nextcraft/apps/ai-service/ai_service/grading/features.py

"""Deterministic process-trace digest (D-028, REQ-3-004).

Pure compute — no LLM, no I/O. `compute_digest` reduces an ordered
TelemetryEvent trace to a compact, bounded `TraceDigest` that is safe to
embed in a grading prompt:

- FIXED fields + small histograms only; NO raw commands, NO file contents,
  NO payloads — the raw trace NEVER reaches the LLM (D-028), which also
  bounds the prompt-injection surface.
- Tolerant to both live trace mixes: daemon-topology traces carry
  `activity` + `file_diff` kinds (workspace watcher), while REPL-driven
  traces carry `command`/`stdin`/`stdout`/`run_result`/`test_result`
  (P2 verification P1). Features derive from whatever kinds are present and
  never crash on absent kinds.

Feature semantics (conservative, deterministic):
- test pass/fail counts + final status derive from `test_result` payloads
  when present, falling back to `run_result` exit codes (0 = pass).
- an error/fix CYCLE = a failing run/test followed by >= 1 edit and then a
  later run/test (pass or fail) — the next observed result closes the cycle.
- idle gaps = wall-clock gaps between consecutive events exceeding
  `idle_threshold_s` (default 120s): count + total seconds.
- command category histogram classifies `command`-kind payloads: build /
  test / file / nav / debug / other.
"""

from __future__ import annotations

from collections import Counter
from typing import TYPE_CHECKING

from pydantic import BaseModel, Field

from ..telemetry.models import TelemetryEvent

if TYPE_CHECKING:  # pragma: no cover - import cycle guard for type checkers
    pass


_IDLE_DEFAULT_S: float = 120.0

_TEST_HINTS = ("test", "pytest", "vitest", "jest", "mocha", "unittest", "go test", "npm test")
_BUILD_HINTS = ("make", "npm run build", "pip install", "pnpm", "cargo build", "gcc", "tsc")
_DEBUG_HINTS = ("gdb", "pdb", "print(", "debug", "strace", "ltrace", "curl", "ping")
_NAV_HINTS = ("ls", "cd", "pwd", "cat ", "grep ", "find", "rg ", "tree", "head", "tail", "less")
_FILE_HINTS = ("mv ", "cp ", "rm ", "mkdir", "touch", "chmod", "nano", "vim", "sed -i", "tee ")


class TraceDigest(BaseModel):
    """Compact, bounded, LLM-safe summary of a process trace (D-028).

    Fixed fields + small histograms. Serializes well under 4 KB; contains no
    raw commands, file contents, or event payloads.
    """

    model_config = {"frozen": True}

    event_count: int = Field(ge=0)
    session_duration_s: float = Field(ge=0.0)
    edit_count: int = Field(ge=0)
    command_count: int = Field(ge=0)
    run_count: int = Field(ge=0)
    test_pass_count: int = Field(ge=0)
    test_fail_count: int = Field(ge=0)
    final_test_status: str = Field(pattern="^(pass|fail|none)$")
    first_test_pass_offset_s: float | None = None
    error_fix_cycles: int = Field(ge=0)
    mean_fix_latency_s: float | None = None
    idle_gap_count: int = Field(ge=0)
    idle_gap_total_s: float = Field(ge=0.0)
    command_categories: dict[str, int] = Field(default_factory=dict)
    kind_histogram: dict[str, int] = Field(default_factory=dict)


def _event_pass_status(event: TelemetryEvent) -> bool | None:
    """True (pass) / False (fail) / None (not a result event) for one event."""
    payload = event.payload or {}
    if event.kind == "test_result":
        if "passed" in payload:
            return bool(payload["passed"])
        if "exit_code" in payload:
            return int(payload["exit_code"]) == 0
        if "status" in payload:
            return str(payload["status"]).lower() in ("pass", "passed", "ok", "success")
        return None
    if event.kind == "run_result":
        if "exit_code" in payload:
            return int(payload["exit_code"]) == 0
        if "ok" in payload:
            return bool(payload["ok"])
        return None
    return None


def _classify_command(text: str) -> str:
    lowered = text.lower()
    if any(h in lowered for h in _TEST_HINTS):
        return "test"
    if any(h in lowered for h in _BUILD_HINTS):
        return "build"
    if any(h in lowered for h in _DEBUG_HINTS):
        return "debug"
    if any(h in lowered for h in _NAV_HINTS):
        return "nav"
    if any(h in lowered for h in _FILE_HINTS):
        return "file"
    return "other"


def _command_text(event: TelemetryEvent) -> str:
    payload = event.payload or {}
    return str(payload.get("cmd") or payload.get("command") or payload.get("line") or "")


def compute_digest(
    trace: list[TelemetryEvent], *, idle_threshold_s: float = _IDLE_DEFAULT_S
) -> TraceDigest:
    """Reduce an ordered trace to a bounded digest. Never raises on odd input."""
    events = sorted(trace, key=lambda e: (e.seq, e.ts))
    if not events:
        return TraceDigest(
            event_count=0,
            session_duration_s=0.0,
            edit_count=0,
            command_count=0,
            run_count=0,
            test_pass_count=0,
            test_fail_count=0,
            final_test_status="none",
            first_test_pass_offset_s=None,
            error_fix_cycles=0,
            mean_fix_latency_s=None,
            idle_gap_count=0,
            idle_gap_total_s=0.0,
            command_categories={},
            kind_histogram={},
        )

    kind_histogram = Counter(e.kind for e in events)
    start_ts = events[0].ts
    end_ts = events[-1].ts
    duration = max(0.0, (end_ts - start_ts).total_seconds())

    edit_count = kind_histogram.get("file_diff", 0)
    command_count = kind_histogram.get("command", 0)
    run_count = kind_histogram.get("run_result", 0)

    # Tests: prefer test_result events; fall back to run_result exit codes.
    test_statuses: list[tuple[TelemetryEvent, bool]] = []
    for e in events:
        if e.kind == "test_result":
            ok = _event_pass_status(e)
            if ok is not None:
                test_statuses.append((e, ok))
    if not test_statuses:
        for e in events:
            if e.kind == "run_result":
                ok = _event_pass_status(e)
                if ok is not None:
                    test_statuses.append((e, ok))

    test_pass_count = sum(1 for _, ok in test_statuses if ok)
    test_fail_count = len(test_statuses) - test_pass_count
    if not test_statuses:
        final_test_status = "none"
    else:
        final_test_status = "pass" if test_statuses[-1][1] else "fail"
    first_pass = next((e for e, ok in test_statuses if ok), None)
    first_pass_offset = (
        max(0.0, (first_pass.ts - start_ts).total_seconds()) if first_pass is not None else None
    )

    # Error/fix cycles: a failing result starts a pending cycle; the NEXT
    # observed result closes it (regardless of outcome) — a fix attempt that
    # fails again is itself another iteration of debugging, so it closes the
    # previous cycle and opens a new one. Edits since the fail mark the
    # close as a genuine fix attempt; latency = first edit -> closing result.
    cycles = 0
    fix_latencies: list[float] = []
    pending_fail_ts: float | None = None  # seconds since start
    edits_since_fail = 0
    first_edit_ts: float | None = None
    for e in events:
        t = max(0.0, (e.ts - start_ts).total_seconds())
        if e.kind == "file_diff":
            if pending_fail_ts is not None:
                if edits_since_fail == 0:
                    first_edit_ts = t
                edits_since_fail += 1
            continue
        ok = _event_pass_status(e)
        if ok is None:
            continue
        if ok is False:
            if pending_fail_ts is not None and edits_since_fail > 0 and first_edit_ts is not None:
                # failed fix attempt: closes the previous cycle, opens a new one
                cycles += 1
                fix_latencies.append(t - first_edit_ts)
            pending_fail_ts = t
            edits_since_fail = 0
            first_edit_ts = None
            continue
        if ok is True and pending_fail_ts is not None:
            if edits_since_fail > 0 and first_edit_ts is not None:
                cycles += 1
                fix_latencies.append(t - first_edit_ts)
            pending_fail_ts = None
            edits_since_fail = 0
            first_edit_ts = None

    mean_fix_latency = (
        sum(fix_latencies) / len(fix_latencies) if fix_latencies else None
    )

    # Idle gaps between consecutive events.
    idle_gap_count = 0
    idle_gap_total = 0.0
    prev_ts = None
    for e in events:
        if prev_ts is not None:
            gap = (e.ts - prev_ts).total_seconds()
            if gap > idle_threshold_s:
                idle_gap_count += 1
                idle_gap_total += gap
        prev_ts = e.ts

    # Command category histogram (command-kind events only).
    categories: Counter[str] = Counter()
    for e in events:
        if e.kind == "command":
            categories[_classify_command(_command_text(e))] += 1

    return TraceDigest(
        event_count=len(events),
        session_duration_s=round(duration, 3),
        edit_count=edit_count,
        command_count=command_count,
        run_count=run_count,
        test_pass_count=test_pass_count,
        test_fail_count=test_fail_count,
        final_test_status=final_test_status,
        first_test_pass_offset_s=(
            round(first_pass_offset, 3) if first_pass_offset is not None else None
        ),
        error_fix_cycles=cycles,
        mean_fix_latency_s=(round(mean_fix_latency, 3) if mean_fix_latency is not None else None),
        idle_gap_count=idle_gap_count,
        idle_gap_total_s=round(idle_gap_total, 3),
        command_categories=dict(sorted(categories.items())),
        kind_histogram=dict(sorted(kind_histogram.items())),
    )