fidelity-ai-workspace/scripts/mattermost-proxy/read-context.py

#!/usr/bin/env python3
"""Read local Mattermost mirror evidence for agent prompts.

The proxy mirror under the active profile inbox is the preferred Mattermost
evidence source when present. This reader gives commands a stable way to load a
small focused view and fall back to the older sync artifacts when the mirror is
not available.
"""

from __future__ import annotations

import argparse
import json
import os
import shlex
import sys
from datetime import date, datetime, timedelta
from pathlib import Path


ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "scripts" / "aiw"))
import profile as aiw_profile  # noqa: E402

DEFAULT_PROFILE = os.getenv("AIW_PROJECT_PROFILE", "fidelity")
MIRROR_DIR = aiw_profile.inbox_dir(DEFAULT_PROFILE, root=ROOT) / "mattermost-mirror"
LEGACY_LATEST = aiw_profile.inbox_dir(DEFAULT_PROFILE, root=ROOT) / "mattermost-latest.md"
LEGACY_GENERATED = ROOT / "scripts" / "mattermost" / "generated" / "mattermost_context.jsonl"
LOCAL_ENV = Path(__file__).resolve().parent / ".env"


def configure_profile_paths(profile: str) -> None:
    global MIRROR_DIR, LEGACY_LATEST
    inbox = aiw_profile.inbox_dir(profile, root=ROOT)
    MIRROR_DIR = inbox / "mattermost-mirror"
    LEGACY_LATEST = inbox / "mattermost-latest.md"


def load_local_env(path: Path = LOCAL_ENV) -> None:
    """Load simple KEY=VALUE pairs from the connector-local .env.

    Existing process environment values win. This keeps the reusable reader
    project-agnostic while allowing each workspace/profile to provide its own
    channel filters without hardcoding them in Python.
    """
    if not path.is_file():
        return
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        if line.startswith("export "):
            line = line[len("export ") :].strip()
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip()
        if not key or key in os.environ:
            continue
        try:
            parsed = shlex.split(value, comments=False, posix=True)
            value = parsed[0] if parsed else ""
        except ValueError:
            value = value.strip('"\'')
        os.environ[key] = value


def previous_workday(today: date) -> date:
    day = today - timedelta(days=1)
    while day.weekday() >= 5:
        day -= timedelta(days=1)
    return day


def daily_by_date_path(day: date) -> Path:
    return MIRROR_DIR / "by-date" / f"{day:%Y}" / f"{day:%m}" / f"{day:%Y-%m-%d}.jsonl"


def print_file(path: Path) -> bool:
    if path.is_file() and path.stat().st_size > 0:
        print(path.read_text(encoding="utf-8"))
        return True
    return False


def read_jsonl(path: Path) -> list[dict]:
    records = []
    if not path.is_file():
        return records
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            records.append(json.loads(line))
        except json.JSONDecodeError:
            continue
    return records


def parse_channels(raw: str | None) -> set[str]:
    if not raw:
        env_raw = os.getenv("AIW_MATTERMOST_CONTEXT_CHANNELS", "") or os.getenv(
            "AIW_MATTERMOST_PROJECT_CHANNELS", ""
        )
        raw = env_raw
    if not raw:
        return set()
    return {item.strip() for item in raw.split(",") if item.strip()}


def filter_channels(records: list[dict], channels: set[str] | None) -> list[dict]:
    if not channels:
        return records
    lowered = {channel.lower() for channel in channels}
    return [
        record
        for record in records
        if str(record.get("channel_name", "")).lower() in lowered
        or str(record.get("channel_id", "")).lower() in lowered
    ]


def trim_records(records: list[dict], limit: int | None) -> list[dict]:
    if limit is None or limit <= 0 or len(records) <= limit:
        return records
    return records[-limit:]


def print_jsonl(records: list[dict]) -> bool:
    if not records:
        return False
    for record in records:
        print(json.dumps(record, ensure_ascii=False, sort_keys=True))
    return True


def fallback() -> None:
    if print_file(LEGACY_LATEST):
        return
    if print_file(LEGACY_GENERATED):
        return
    print("No Mattermost context available.")


def mode_latest() -> None:
    print("## Mattermost mirror latest context")
    if print_file(MIRROR_DIR / "latest.md"):
        return
    records = read_jsonl(MIRROR_DIR / "latest.jsonl")
    if print_jsonl(records):
        return
    print("No proxy mirror latest context available; checking configured sync artifacts.")
    fallback()


def print_records_section(title: str, records: list[dict], limit: int | None = None) -> bool:
    print(title)
    records = trim_records(records, limit)
    if print_jsonl(records):
        return True
    print("No matching Mattermost mirror records.")
    return False


def mode_previous_workday(today_raw: str | None, channels: set[str] | None = None, limit: int | None = None) -> None:
    today = datetime.strptime(today_raw, "%Y-%m-%d").date() if today_raw else date.today()
    day = previous_workday(today)
    path = daily_by_date_path(day)
    records = filter_channels(read_jsonl(path), channels)
    if print_records_section(f"## Mattermost mirror previous-workday context ({day.isoformat()})", records, limit):
        return
    if channels:
        print("Filtered to project channels: " + ", ".join(sorted(channels)))
    print("No proxy mirror previous-workday project context available; not falling back to broad latest context.")


def mode_standup(today_raw: str | None, channels: set[str], limit: int | None) -> None:
    """Print a compact standup-focused view.

    Standup mode intentionally avoids latest.md because latest.md is a bounded
    global window and can contain stale or unrelated channels. Use date-bucketed
    mirror files filtered to known project channels instead.
    """
    today = datetime.strptime(today_raw, "%Y-%m-%d").date() if today_raw else date.today()
    day = previous_workday(today)
    previous_records = filter_channels(read_jsonl(daily_by_date_path(day)), channels)
    today_records = filter_channels(read_jsonl(daily_by_date_path(today)), channels)

    print("## Mattermost mirror standup context")
    if channels:
        print("Filtered to configured context channels: " + ", ".join(sorted(channels)))
    else:
        print("No context channel filter configured; using all mirrored date-bucket records.")
    print_records_section(f"\n### Previous workday ({day.isoformat()})", previous_records, limit)
    print_records_section(f"\n### Today so far ({today.isoformat()})", today_records, limit)


def mode_focused() -> None:
    records = read_jsonl(MIRROR_DIR / "latest.jsonl")
    if not records:
        records = read_jsonl(LEGACY_GENERATED)
    if not records:
        print("No Mattermost context available.")
        return

    manager_names = {"jeff", "jeff.dewitte"}
    manager_records = [record for record in records if str(record.get("username", "")).lower() in manager_names]
    focused = manager_records[-10:] if manager_records else records[-15:]
    print_jsonl(focused)


def main() -> None:
    load_local_env()

    parser = argparse.ArgumentParser()
    parser.add_argument("--profile", default=DEFAULT_PROFILE)
    parser.add_argument("--mode", choices=["latest", "previous-workday", "standup", "focused"], default="latest")
    parser.add_argument("--today", default="")
    parser.add_argument(
        "--channels",
        default="",
        help="Comma-separated channel names or IDs. Defaults to AIW_MATTERMOST_CONTEXT_CHANNELS from environment/.env when set.",
    )
    parser.add_argument("--limit", type=int, default=80, help="Max records per section; use 0 for no limit.")
    args = parser.parse_args()
    configure_profile_paths(args.profile)
    channels = parse_channels(args.channels or None)
    limit = args.limit if args.limit > 0 else None

    if args.mode == "latest":
        mode_latest()
    elif args.mode == "previous-workday":
        mode_previous_workday(args.today or None, channels=None, limit=limit)
    elif args.mode == "standup":
        mode_standup(args.today or None, channels, limit)
    elif args.mode == "focused":
        mode_focused()


if __name__ == "__main__":
    main()