From 71188c56d2770a2ad4df4348e75c1be909aa2fe0 Mon Sep 17 00:00:00 2001 From: aaedavel <{akash.aedavelli@gmail.com}> Date: Tue, 28 Apr 2026 14:57:29 -0500 Subject: [PATCH] feat(investigations): add audited lifecycle tools Add durable Core investigation storage and MCP lifecycle/read APIs so Hermes can run Slice 9 loops while Core keeps digest-only audit records and render events. --- HANDOFF.md | 2 +- ...026-04-19-slice9-agentic-investigations.md | 21 +- ...28-slice9-investigation-render-contract.md | 58 +- minx_mcp/core/investigations.py | 793 ++++++++++++++++++ minx_mcp/core/server.py | 2 + minx_mcp/core/tools/investigations.py | 278 ++++++ .../schema/migrations/027_investigations.sql | 36 + tests/test_core_server.py | 6 + tests/test_investigations.py | 272 ++++++ tests/test_memory_service.py | 3 +- 10 files changed, 1449 insertions(+), 22 deletions(-) create mode 100644 minx_mcp/core/investigations.py create mode 100644 minx_mcp/core/tools/investigations.py create mode 100644 minx_mcp/schema/migrations/027_investigations.sql create mode 100644 tests/test_investigations.py diff --git a/HANDOFF.md b/HANDOFF.md index 8ba0d40..7954e1b 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -208,7 +208,7 @@ Each PR ships independently green (`ruff` + `mypy minx_mcp` + `pytest`), with mi - One spec doc under `docs/superpowers/specs/YYYY-MM-DD-slice6X-*.md` (adversarially reviewed before implementation). - One plan doc under `docs/superpowers/plans/YYYY-MM-DD-slice6X-*.md` (step-by-step execution checklist). -- Sequentially numbered migration (next filename: `027_*.sql` after `026_memory_capture_fts.sql`). Slice 9 investigations should claim the next available migration. +- Sequentially numbered migration (next filename: `028_*.sql` after `027_investigations.sql`). - Implemented-slices row appended to the table above, with LOC / date / verification block. - Operator post-upgrade step added under "Post-Upgrade Operator Steps" if the migration is not fully reversible from application-level data (6g needed a backfill; 6i will need a one-shot FTS5 rebuild; 6l will need a one-shot embedding backfill and a cost-ceiling env var). - After deploying `026_memory_capture_fts.sql`, run `python -m scripts.rebuild_memory_fts /path/to/minx.db` so any pre-existing `captured_thought` rows are indexed by `payload.text` and `payload.capture_type`. diff --git a/docs/superpowers/specs/2026-04-19-slice9-agentic-investigations.md b/docs/superpowers/specs/2026-04-19-slice9-agentic-investigations.md index 8a18eac..aa6ca65 100644 --- a/docs/superpowers/specs/2026-04-19-slice9-agentic-investigations.md +++ b/docs/superpowers/specs/2026-04-19-slice9-agentic-investigations.md @@ -46,9 +46,9 @@ Reasons: | Surface | Why agentic | Indicative trajectory | | --------------------------------- | ---------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------- | -| `minx_investigate(question)` | Causal/exploratory questions. LLM decides whether to drill into merchants, categories, meals, goals. | `finance_categories` → `finance_transactions(...)` → maybe `meals_list` → maybe `get_insight_history` → compose | +| `minx_investigate(question)` | Causal/exploratory questions. LLM decides whether to drill into merchants, categories, meals, goals. | `finance_query` → maybe `meals_list` → maybe `get_insight_history` → compose | | `minx_plan(objective)` | Scheduling/planning across domains. Depends on what it finds. | `goal_list` → `get_goal_trajectory` → `training_list` → `meals_list` → draft → revise | -| `minx_retro(period, subject)` | Causal analysis across months. LLM picks which detectors to replay, which transactions to sample. | `get_insight_history` → `goal_trajectory` → sampling tools → synthesize | +| `minx_retro(period, subject)` | Causal analysis across months. LLM picks which detectors to replay, which transactions to sample. | `get_insight_history` → `get_goal_trajectory` → sampling tools → synthesize | | `minx_onboard_entity(kind, name)` | Hydrates an entity/pattern page from scratch. Branches on what it finds. | `finance_transactions(merchant=...)` → `memory_list(subject=...)` → maybe `persist_note` | @@ -56,7 +56,7 @@ Common shape: **one question in, one report out, unpredictable middle.** ## 5) Schema (Core) -Migration filename: use the next available sequential migration when this slice lands. As of 2026-04-27, Slice 6i-6l are expected to ship before investigations, so this spec no longer pre-claims `021_investigations.sql`. +Migration filename: use the next available sequential migration when this slice lands. After `026_memory_capture_fts.sql`, this should be `027_investigations.sql` unless another migration lands first. ```sql CREATE TABLE investigations ( @@ -86,7 +86,7 @@ CREATE INDEX idx_investigations_kind_started ON investigations(kind, started_at CREATE INDEX idx_investigations_running ON investigations(status) WHERE status = 'running'; ``` -**Trajectory storage policy:** `trajectory_json` stores a **digest** per step (tool name, arg hash, result row count / bytes, latency). It does NOT store full tool outputs — those can be large and contain PII. Full outputs are reconstructable by replaying the tools against the DB at investigation time. +**Trajectory storage policy:** `trajectory_json` stores a **digest** per step (tool name, arg hash, result row count / bytes, latency). It does NOT store full tool outputs — those can be large and contain PII. Some outputs may be approximately reproducible by re-querying domain tools, but replay is not a durable audit guarantee because data, code, and time-dependent results can change. **Render storage policy:** `response_template` and `response_slots_json` store the latest lifecycle event so read APIs can expose a stable render surface without parsing trajectory text. Step-level render events are stored inside `trajectory_json` step entries. @@ -103,16 +103,18 @@ complete_investigation( investigation_id, status, # 'succeeded' | 'failed' | 'cancelled' | 'budget_exhausted' answer_md, + citation_refs, # optional list of typed references used by the harness answer tool_call_count, token_input, token_output, cost_usd, error_message, ) -> {"investigation_id": int, "response_template": "investigation.completed|investigation.failed|investigation.cancelled|investigation.budget_exhausted", "response_slots": {...}} -log_investigation(...) # convenience wrapper with the same logging role as log_playbook_run; - # MCP return shape follows the render-contract amendment +log_investigation(kind, question, context_json, harness, trajectory_json, status, answer_md, citation_refs, ...) + # convenience wrapper with the same logging role as log_playbook_run; + # MCP return shape follows the render-contract amendment -investigation_history(kind=None, since=None, days=30, limit=100) -> {"runs": [...], "truncated": bool} +investigation_history(kind=None, harness=None, status=None, since=None, days=30, limit=100) -> {"runs": [...], "truncated": bool} investigation_get(investigation_id) -> {"run": {...}} # includes trajectory and latest response_template/response_slots ``` @@ -180,7 +182,8 @@ Ship order: 9a → 9b → 9d (first usable surface) → 9c + 9e + 9f + 9g in any - Concurrent starts with different kinds don't collide; same-kind concurrent is allowed (investigations are user-initiated, no cron contention). - `append_investigation_step` rejects steps after terminal status. - Trajectory digest: `result_digest` never contains raw tool output bytes; `context_json` goes through a redaction pass for known PII fields (email, phone, account numbers). -- `investigation_history` pagination/filter matches `playbook_history` semantics. +- `investigation_history` pagination/filter matches `playbook_history` semantics for `kind`, `harness`, `status`, `since`, `days`, and `limit`. +- `complete_investigation` and `log_investigation` persist typed `citation_refs` separately from `answer_md`. ### Harness tests (9d–9f, outside this repo) @@ -190,6 +193,6 @@ Ship order: 9a → 9b → 9d (first usable surface) → 9c + 9e + 9f + 9g in any ## 11) Relationship to Other Slices -- **Slice 6 (Memory):** investigations can cite memories by id in `answer_md`; `memory_get`, `memory_list`, FTS5 search, memory graph edges, and embeddings/hybrid retrieval are primary inputs. +- **Slice 6 (Memory):** investigations can cite memories by id in structured `citation_refs`; `memory_get`, `memory_list`, FTS5 search, memory graph edges, and embeddings/hybrid retrieval are primary inputs. - **Slice 8 (Playbooks):** audit pattern (two-phase + convenience wrapper) is lifted directly. A sibling to `playbook_reconcile_crashed` can be added later if crashed-running investigations need automated reconciliation; specify that tool explicitly before shipping it. - **Slice 5 (Harness Adaptation):** a second harness would reimplement the loop against the same Core API. The agent-loop pattern is harness-specific; the tool surface is portable. \ No newline at end of file diff --git a/docs/superpowers/specs/2026-04-28-slice9-investigation-render-contract.md b/docs/superpowers/specs/2026-04-28-slice9-investigation-render-contract.md index 57fde82..44a4b50 100644 --- a/docs/superpowers/specs/2026-04-28-slice9-investigation-render-contract.md +++ b/docs/superpowers/specs/2026-04-28-slice9-investigation-render-contract.md @@ -72,6 +72,7 @@ Completion should return: "kind": "investigate", "status": "succeeded", "tool_call_count": 8, + "citation_count": 4, "cited_memory_count": 3, "cost_usd": 0.12 } @@ -80,6 +81,8 @@ Completion should return: Hermes may render those as "Investigation complete" or a richer explanation, but the final prose lives outside Core. +`complete_investigation` and `log_investigation` should accept optional `citation_refs`, stored in `citation_refs_json`, so references used by `answer_md` are available without parsing prose. + ## Schema Adjustments The existing Slice 9 spec stores `answer_md`. Under this update: @@ -101,7 +104,18 @@ Where: - `response_template` stores the latest lifecycle/event template key. - `response_slots_json` stores JSON slots for that latest event. -- `citation_refs_json` stores references such as memory ids, investigation ids, tool result digests, or vault paths used by the harness answer. +- `citation_refs_json` stores references used by the harness answer. It is a JSON list of typed objects: + +```json +[ + {"type": "memory", "id": 123}, + {"type": "investigation", "id": 42}, + {"type": "vault_path", "path": "Minx/Reviews/2026-04-28.md"}, + {"type": "tool_result_digest", "tool": "finance_query", "digest": "9a2f1c4e7b8d..."} +] +``` + +Allowed reference `type` values are `memory`, `investigation`, `vault_path`, and `tool_result_digest`. Unknown reference types should be rejected until a follow-up spec defines them. Use these columns for the initial Core implementation so `investigation_history` and `investigation_get` have a stable latest-event surface. Step-level events still live in `trajectory_json` entries. A trajectory-only storage approach should be a deliberate later simplification, and only if history/get tools continue exposing `response_template`, `response_slots`, and step event fields without parsing prose. @@ -114,21 +128,31 @@ Use these columns for the initial Core implementation so `investigation_history` "step": 3, "event_template": "investigation.step_logged", "event_slots": { - "tool": "finance_query", - "result_digest": "sha256:9a2f1c4e7b8d", - "latency_ms": 182, "row_count": 12 }, "tool": "finance_query", - "args_digest": "sha256:6f12b4c8d901", - "result_digest": "sha256:9a2f1c4e7b8d", + "args_digest": "6f12b4c8d901...", + "result_digest": "9a2f1c4e7b8d...", "latency_ms": 182 } ``` Required fields: `step`, `event_template`, `event_slots`, `tool`, `args_digest`, `result_digest`, and `latency_ms`. -Allowed `event_slots` values are structured digests, counts, enum-like labels, ids, booleans, numbers, and short normalized strings. No raw tool output should be stored in `event_slots`; use `result_digest`, `row_count`, `byte_count`, or citation ids instead. +Digest fields are raw lowercase SHA-256 hex strings over canonical JSON or canonical text, without a `sha256:` prefix. This matches the existing Core fingerprint helper style. + +Allowed `event_slots` values are structured digests, counts, enum-like labels, ids, booleans, numbers, and short normalized strings. No raw tool output should be stored in `event_slots`; use `result_digest`, `row_count`, `byte_count`, or citation ids instead. `event_slots` may include render-relevant summaries, but it does not need to duplicate top-level `tool`, `args_digest`, `result_digest`, or `latency_ms`. + +Validation rules: + +- `step` must be a positive integer. +- `event_template` must be one of the investigation template keys listed above. +- `tool` must be a non-empty normalized tool name. +- `args_digest` and `result_digest` must match `[0-9a-f]{64}`. +- `latency_ms` must be a non-negative integer. +- `event_slots` must be a JSON object with at most 32 top-level keys, max nesting depth 4, and string leaves capped at 1024 UTF-8 bytes. +- The serialized `step_json` must be capped at 16 KiB. +- Reject raw-output keys such as `raw_output`, `tool_output`, `result_json`, `result_rows`, `transcript`, and `messages`. ## Confirmations @@ -146,7 +170,16 @@ If an investigation needs user confirmation before a risky step, Core should sto } ``` -`append_investigation_step` may return `response_template == "investigation.needs_confirmation"` instead of `investigation.step_logged` when the appended step records a proposed risky action that is waiting on the user. Hermes renders the prompt and records the user's decision by calling the appropriate Core/domain tool. Core should not invent the confirmation wording. +`append_investigation_step` returns `response_template == "investigation.needs_confirmation"` only when the appended step's `event_template` is exactly `investigation.needs_confirmation`; otherwise it returns the step's render event, normally `investigation.step_logged`. The investigation status remains `running`. Hermes renders the prompt, records the user's decision by calling the appropriate Core/domain tool, and should append a later `investigation.step_logged` step that records the decision digest. Core should not invent the confirmation wording or treat confirmation slots as authorization for a domain mutation. + +## Redaction And Blocking + +Core should apply deterministic local secret handling before persistence: + +- Redact redactable secret-shaped values in `question`, `answer_md`, `error_message`, JSON string leaves in `context_json`, JSON string leaves in `citation_refs`, and `event_slots`. +- Block non-redactable secret-shaped values, such as private key blocks, with `INVALID_INPUT`. +- Do not scan digest fields as secrets. +- Preserve structured fields when redacting JSON leaves; do not collapse structured objects into prose. ## Testing @@ -155,7 +188,10 @@ Core tests should cover: - `start_investigation` returns `response_template == "investigation.started"`. - `append_investigation_step` stores step `event_template` and JSON-safe `event_slots`. - `complete_investigation` returns `investigation.completed`, `investigation.failed`, `investigation.cancelled`, or `investigation.budget_exhausted` based on status. -- Stored slots never include raw tool output. +- Stored slots never include raw tool output and reject raw-output key names. +- Digest fields reject non-hex or prefixed digest strings. +- `citation_refs` accept only the typed reference schema and are exposed by history/get without prose parsing. +- `question`, `context_json`, `answer_md`, `error_message`, `citation_refs`, and `event_slots` follow the redaction/blocking policy. - `answer_md`, if present, is accepted as harness-authored content and not generated by Core. - history/get tools expose structured event/template data without requiring prose parsing. @@ -167,7 +203,7 @@ Hermes tests, outside this repo, should cover: ## Read Surfaces -`investigation_history` should expose the latest lifecycle render event for each run: +`investigation_history(kind=None, harness=None, status=None, since=None, days=30, limit=100)` should expose the latest lifecycle render event for each run: ```json { @@ -188,7 +224,7 @@ Hermes tests, outside this repo, should cover: } ``` -`investigation_get` should include the same latest `response_template` / `response_slots` plus the trajectory entries with each step's `event_template` / `event_slots`. `log_investigation` is a convenience wrapper over the same lifecycle storage rules; it should return the terminal lifecycle render hint that matches the logged status. +`investigation_get` should include the same latest `response_template` / `response_slots`, `citation_refs`, and the trajectory entries with each step's `event_template` / `event_slots`. `log_investigation` is a convenience wrapper over the same lifecycle storage rules; it should return the terminal lifecycle render hint that matches the logged status. ## Relationship To The Existing Slice 9 Spec diff --git a/minx_mcp/core/investigations.py b/minx_mcp/core/investigations.py new file mode 100644 index 0000000..5f1b5b4 --- /dev/null +++ b/minx_mcp/core/investigations.py @@ -0,0 +1,793 @@ +"""Slice 9 investigation lifecycle helpers.""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from datetime import UTC, datetime, timedelta +from sqlite3 import Connection, Row +from typing import Any + +from minx_mcp.contracts import ConflictError, InvalidInputError, NotFoundError +from minx_mcp.core.secret_scanner import SecretVerdictKind, redact_secrets +from minx_mcp.time_utils import utc_now_isoformat + +KIND_VALUES = frozenset({"investigate", "plan", "retro", "onboard", "other"}) +TERMINAL_STATUSES = frozenset({"succeeded", "failed", "cancelled", "budget_exhausted"}) +ALL_STATUSES = TERMINAL_STATUSES | {"running"} +STEP_EVENT_TEMPLATES = frozenset({"investigation.step_logged", "investigation.needs_confirmation"}) +TERMINAL_RESPONSE_TEMPLATES = { + "succeeded": "investigation.completed", + "failed": "investigation.failed", + "cancelled": "investigation.cancelled", + "budget_exhausted": "investigation.budget_exhausted", +} +MAX_HISTORY_LIMIT = 1000 +MAX_JSON_DEPTH = 4 +MAX_TOP_LEVEL_KEYS = 32 +MAX_STRING_BYTES = 1024 +MAX_STEP_BYTES = 16 * 1024 +_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$") +_TOOL_NAME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.:-]{0,127}$") +_RAW_OUTPUT_KEYS = frozenset( + { + "raw_output", + "tool_output", + "result_json", + "result_rows", + "transcript", + "messages", + } +) + + +def canonical_json_digest(value: Any) -> str: + """Return a raw lowercase SHA-256 hex digest for canonical JSON.""" + normalized = _normalize_json_value(value, field_name="value", redact=False) + payload = json.dumps(normalized, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def start_investigation( + conn: Connection, + *, + kind: str, + question: str, + context_json: dict[str, Any] | None, + harness: str, +) -> dict[str, object]: + normalized_kind = _normalize_kind(kind) + normalized_harness = _require_non_empty(harness, "harness") + normalized_question = _redact_text(_require_non_empty(question, "question"), "question") + normalized_context = _normalize_json_object(context_json or {}, field_name="context_json", redact=True) + now = utc_now_isoformat() + response_slots: dict[str, object] = { + "kind": normalized_kind, + "harness": normalized_harness, + "status": "running", + } + + conn.execute("BEGIN IMMEDIATE") + try: + cur = conn.execute( + """ + INSERT INTO investigations ( + harness, + kind, + question, + context_json, + status, + trajectory_json, + response_template, + response_slots_json, + citation_refs_json, + started_at + ) VALUES (?, ?, ?, ?, 'running', '[]', 'investigation.started', ?, '[]', ?) + """, + ( + normalized_harness, + normalized_kind, + normalized_question, + _dump_json(normalized_context), + _dump_json(response_slots), + now, + ), + ) + if cur.lastrowid is None: + raise RuntimeError("investigations insert did not return a row id") + investigation_id = int(cur.lastrowid) + response_slots["investigation_id"] = investigation_id + conn.execute( + "UPDATE investigations SET response_slots_json = ? WHERE id = ?", + (_dump_json(response_slots), investigation_id), + ) + result = { + "investigation_id": investigation_id, + "response_template": "investigation.started", + "response_slots": response_slots, + } + conn.commit() + except Exception: + if conn.in_transaction: + conn.rollback() + raise + return result + + +def append_investigation_step( + conn: Connection, + *, + investigation_id: int, + step_json: dict[str, Any], +) -> dict[str, object]: + normalized_id = _normalize_positive_int(investigation_id, "investigation_id") + normalized_step = normalize_step_json(step_json) + + conn.execute("BEGIN IMMEDIATE") + try: + row = conn.execute( + "SELECT id, kind, harness, status, trajectory_json FROM investigations WHERE id = ?", + (normalized_id,), + ).fetchone() + if row is None: + raise NotFoundError(f"investigation {normalized_id} not found") + if str(row["status"]) != "running": + raise ConflictError( + f"investigation {normalized_id} is not running", + data={"investigation_id": normalized_id, "status": str(row["status"])}, + ) + trajectory = _json_loads_list(row["trajectory_json"], "trajectory_json") + trajectory.append(normalized_step) + response_template = str(normalized_step["event_template"]) + response_slots: dict[str, object] = dict(normalized_step["event_slots"]) + response_slots.update( + { + "investigation_id": normalized_id, + "kind": str(row["kind"]), + "harness": str(row["harness"]), + "status": "running", + "step": int(normalized_step["step"]), + "tool": str(normalized_step["tool"]), + } + ) + cur = conn.execute( + """ + UPDATE investigations + SET trajectory_json = ?, + response_template = ?, + response_slots_json = ? + WHERE id = ? AND status = 'running' + """, + (_dump_json(trajectory), response_template, _dump_json(response_slots), normalized_id), + ) + if int(cur.rowcount or 0) != 1: + raise ConflictError( + f"investigation {normalized_id} changed concurrently", + data={"investigation_id": normalized_id}, + ) + result = {"ok": True, "response_template": response_template, "response_slots": response_slots} + conn.commit() + except Exception: + if conn.in_transaction: + conn.rollback() + raise + return result + + +def complete_investigation( + conn: Connection, + *, + investigation_id: int, + status: str, + answer_md: str | None, + citation_refs: list[dict[str, Any]] | None, + tool_call_count: int | None, + token_input: int | None, + token_output: int | None, + cost_usd: float | None, + error_message: str | None, +) -> dict[str, object]: + normalized_id = _normalize_positive_int(investigation_id, "investigation_id") + normalized_status = _normalize_terminal_status(status) + normalized_answer = _normalize_optional_redacted_text(answer_md, "answer_md") + normalized_citations = normalize_citation_refs(citation_refs or []) + normalized_tool_calls = _normalize_optional_non_negative_int(tool_call_count, "tool_call_count") + normalized_token_input = _normalize_optional_non_negative_int(token_input, "token_input") + normalized_token_output = _normalize_optional_non_negative_int(token_output, "token_output") + normalized_cost = _normalize_optional_non_negative_float(cost_usd, "cost_usd") + normalized_error = _normalize_optional_redacted_text(error_message, "error_message") + completed_at = utc_now_isoformat() + + conn.execute("BEGIN IMMEDIATE") + try: + row = conn.execute( + "SELECT id, kind, harness, status FROM investigations WHERE id = ?", + (normalized_id,), + ).fetchone() + if row is None: + raise NotFoundError(f"investigation {normalized_id} not found") + if str(row["status"]) != "running": + raise ConflictError( + f"investigation {normalized_id} is not running", + data={"investigation_id": normalized_id, "status": str(row["status"])}, + ) + response_template = TERMINAL_RESPONSE_TEMPLATES[normalized_status] + response_slots = _terminal_response_slots( + row, + status=normalized_status, + citation_refs=normalized_citations, + tool_call_count=normalized_tool_calls, + cost_usd=normalized_cost, + ) + cur = conn.execute( + """ + UPDATE investigations + SET status = ?, + answer_md = ?, + citation_refs_json = ?, + tool_call_count = ?, + token_input = ?, + token_output = ?, + cost_usd = ?, + error_message = ?, + completed_at = ?, + response_template = ?, + response_slots_json = ? + WHERE id = ? AND status = 'running' + """, + ( + normalized_status, + normalized_answer, + _dump_json(normalized_citations), + normalized_tool_calls, + normalized_token_input, + normalized_token_output, + normalized_cost, + normalized_error, + completed_at, + response_template, + _dump_json(response_slots), + normalized_id, + ), + ) + if int(cur.rowcount or 0) != 1: + raise ConflictError( + f"investigation {normalized_id} changed concurrently", + data={"investigation_id": normalized_id}, + ) + result = { + "investigation_id": normalized_id, + "response_template": response_template, + "response_slots": response_slots, + } + conn.commit() + except Exception: + if conn.in_transaction: + conn.rollback() + raise + return result + + +def log_investigation( + conn: Connection, + *, + kind: str, + question: str, + context_json: dict[str, Any] | None, + harness: str, + trajectory_json: list[dict[str, Any]] | None, + status: str, + answer_md: str | None, + citation_refs: list[dict[str, Any]] | None, + tool_call_count: int | None, + token_input: int | None, + token_output: int | None, + cost_usd: float | None, + error_message: str | None, +) -> dict[str, object]: + normalized_kind = _normalize_kind(kind) + normalized_harness = _require_non_empty(harness, "harness") + normalized_question = _redact_text(_require_non_empty(question, "question"), "question") + normalized_context = _normalize_json_object(context_json or {}, field_name="context_json", redact=True) + normalized_trajectory = [normalize_step_json(step) for step in (trajectory_json or [])] + normalized_status = _normalize_terminal_status(status) + normalized_answer = _normalize_optional_redacted_text(answer_md, "answer_md") + normalized_citations = normalize_citation_refs(citation_refs or []) + normalized_tool_calls = _normalize_optional_non_negative_int(tool_call_count, "tool_call_count") + normalized_token_input = _normalize_optional_non_negative_int(token_input, "token_input") + normalized_token_output = _normalize_optional_non_negative_int(token_output, "token_output") + normalized_cost = _normalize_optional_non_negative_float(cost_usd, "cost_usd") + normalized_error = _normalize_optional_redacted_text(error_message, "error_message") + now = utc_now_isoformat() + response_template = TERMINAL_RESPONSE_TEMPLATES[normalized_status] + + conn.execute("BEGIN IMMEDIATE") + try: + cur = conn.execute( + """ + INSERT INTO investigations ( + harness, + kind, + question, + context_json, + status, + answer_md, + trajectory_json, + response_template, + citation_refs_json, + tool_call_count, + token_input, + token_output, + cost_usd, + started_at, + completed_at, + error_message + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + normalized_harness, + normalized_kind, + normalized_question, + _dump_json(normalized_context), + normalized_status, + normalized_answer, + _dump_json(normalized_trajectory), + response_template, + _dump_json(normalized_citations), + normalized_tool_calls, + normalized_token_input, + normalized_token_output, + normalized_cost, + now, + now, + normalized_error, + ), + ) + if cur.lastrowid is None: + raise RuntimeError("investigations insert did not return a row id") + investigation_id = int(cur.lastrowid) + row = conn.execute( + "SELECT id, kind, harness, status FROM investigations WHERE id = ?", + (investigation_id,), + ).fetchone() + if row is None: + raise RuntimeError("inserted investigation row was not readable") + response_slots = _terminal_response_slots( + row, + status=normalized_status, + citation_refs=normalized_citations, + tool_call_count=normalized_tool_calls, + cost_usd=normalized_cost, + ) + conn.execute( + "UPDATE investigations SET response_slots_json = ? WHERE id = ?", + (_dump_json(response_slots), investigation_id), + ) + result = { + "investigation_id": investigation_id, + "response_template": response_template, + "response_slots": response_slots, + } + conn.commit() + except Exception: + if conn.in_transaction: + conn.rollback() + raise + return result + + +def investigation_history( + conn: Connection, + *, + kind: str | None, + harness: str | None, + status: str | None, + since: str | None, + days: int, + limit: int, +) -> dict[str, object]: + normalized_kind = _normalize_optional_kind(kind) + normalized_harness = _normalize_optional_text(harness) + normalized_status = _normalize_optional_status(status) + cutoff = _resolve_history_cutoff(since, days) + normalized_limit = _normalize_limit(limit) + rows = conn.execute( + """ + SELECT * + FROM investigations + WHERE started_at >= ? + AND (? IS NULL OR kind = ?) + AND (? IS NULL OR harness = ?) + AND (? IS NULL OR status = ?) + ORDER BY started_at DESC, id DESC + LIMIT ? + """, + ( + cutoff, + normalized_kind, + normalized_kind, + normalized_harness, + normalized_harness, + normalized_status, + normalized_status, + normalized_limit + 1, + ), + ).fetchall() + return { + "runs": [_row_as_summary(row) for row in rows[:normalized_limit]], + "truncated": len(rows) > normalized_limit, + } + + +def investigation_get(conn: Connection, *, investigation_id: int) -> dict[str, object]: + normalized_id = _normalize_positive_int(investigation_id, "investigation_id") + row = conn.execute("SELECT * FROM investigations WHERE id = ?", (normalized_id,)).fetchone() + if row is None: + raise NotFoundError(f"investigation {normalized_id} not found") + return {"run": _row_as_detail(row)} + + +def recent_resource_payload(conn: Connection, *, limit: int = 20) -> dict[str, object]: + return investigation_history(conn, kind=None, harness=None, status=None, since=None, days=30, limit=limit) + + +def investigation_resource_payload(conn: Connection, *, investigation_id: int) -> dict[str, object]: + return investigation_get(conn, investigation_id=investigation_id) + + +def normalize_step_json(step_json: dict[str, Any]) -> dict[str, Any]: + if not isinstance(step_json, dict): + raise InvalidInputError("step_json must be a JSON object") + _reject_raw_output_keys(step_json, "step_json") + serialized_step = json.dumps( + step_json, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + if len(serialized_step) > MAX_STEP_BYTES: + raise InvalidInputError("step_json is too large") + required = {"step", "event_template", "event_slots", "tool", "args_digest", "result_digest", "latency_ms"} + missing = required - set(step_json) + if missing: + raise InvalidInputError(f"step_json missing required fields: {', '.join(sorted(missing))}") + event_template = _require_non_empty(str(step_json["event_template"]), "event_template") + if event_template not in STEP_EVENT_TEMPLATES: + allowed = ", ".join(sorted(STEP_EVENT_TEMPLATES)) + raise InvalidInputError(f"event_template must be one of: {allowed}") + tool = _normalize_tool_name(step_json["tool"]) + return { + **{ + key: _normalize_json_value(value, field_name=f"step_json.{key}", redact=True) + for key, value in step_json.items() + if key not in required + }, + "step": _normalize_positive_int(step_json["step"], "step"), + "event_template": event_template, + "event_slots": _normalize_json_object(step_json["event_slots"], field_name="event_slots", redact=True), + "tool": tool, + "args_digest": _normalize_digest(step_json["args_digest"], "args_digest"), + "result_digest": _normalize_digest(step_json["result_digest"], "result_digest"), + "latency_ms": _normalize_non_negative_int(step_json["latency_ms"], "latency_ms"), + } + + +def normalize_citation_refs(citation_refs: list[dict[str, Any]]) -> list[dict[str, Any]]: + if not isinstance(citation_refs, list): + raise InvalidInputError("citation_refs must be a list") + normalized: list[dict[str, Any]] = [] + for index, item in enumerate(citation_refs): + if not isinstance(item, dict): + raise InvalidInputError("citation_refs entries must be objects") + ref_type = _require_non_empty(str(item.get("type", "")), f"citation_refs[{index}].type") + if ref_type == "memory" or ref_type == "investigation": + _require_keys(item, {"type", "id"}, f"citation_refs[{index}]") + normalized.append( + { + "type": ref_type, + "id": _normalize_positive_int(item["id"], f"citation_refs[{index}].id"), + } + ) + elif ref_type == "vault_path": + _require_keys(item, {"type", "path"}, f"citation_refs[{index}]") + normalized.append( + { + "type": ref_type, + "path": _redact_text(_require_non_empty(str(item["path"]), "path"), "path"), + } + ) + elif ref_type == "tool_result_digest": + _require_keys(item, {"type", "tool", "digest"}, f"citation_refs[{index}]") + normalized.append( + { + "type": ref_type, + "tool": _normalize_tool_name(item["tool"]), + "digest": _normalize_digest(item["digest"], "digest"), + } + ) + else: + raise InvalidInputError( + "citation_refs type must be one of: investigation, memory, tool_result_digest, vault_path" + ) + return normalized + + +def _row_as_summary(row: Row) -> dict[str, object]: + return { + "investigation_id": int(row["id"]), + "kind": str(row["kind"]), + "harness": str(row["harness"]), + "status": str(row["status"]), + "started_at": str(row["started_at"]), + "completed_at": row["completed_at"], + "response_template": row["response_template"], + "response_slots": _json_loads_dict(row["response_slots_json"], "response_slots_json"), + } + + +def _row_as_detail(row: Row) -> dict[str, object]: + data = _row_as_summary(row) + data.update( + { + "question": str(row["question"]), + "context_json": _json_loads_dict(row["context_json"], "context_json"), + "answer_md": row["answer_md"], + "trajectory": _json_loads_list(row["trajectory_json"], "trajectory_json"), + "citation_refs": _json_loads_list(row["citation_refs_json"], "citation_refs_json"), + "tool_call_count": row["tool_call_count"], + "token_input": row["token_input"], + "token_output": row["token_output"], + "cost_usd": row["cost_usd"], + "error_message": row["error_message"], + } + ) + return data + + +def _terminal_response_slots( + row: Row, + *, + status: str, + citation_refs: list[dict[str, Any]], + tool_call_count: int | None, + cost_usd: float | None, +) -> dict[str, object]: + return { + "investigation_id": int(row["id"]), + "kind": str(row["kind"]), + "harness": str(row["harness"]), + "status": status, + "tool_call_count": tool_call_count, + "cost_usd": cost_usd, + "citation_count": len(citation_refs), + "cited_memory_count": sum(1 for ref in citation_refs if ref.get("type") == "memory"), + } + + +def _normalize_kind(kind: str) -> str: + normalized = _require_non_empty(kind, "kind").lower() + if normalized not in KIND_VALUES: + allowed = ", ".join(sorted(KIND_VALUES)) + raise InvalidInputError(f"kind must be one of: {allowed}") + return normalized + + +def _normalize_optional_kind(kind: str | None) -> str | None: + normalized = _normalize_optional_text(kind) + return None if normalized is None else _normalize_kind(normalized) + + +def _normalize_terminal_status(status: str) -> str: + normalized = _require_non_empty(status, "status").lower() + if normalized not in TERMINAL_STATUSES: + allowed = ", ".join(sorted(TERMINAL_STATUSES)) + raise InvalidInputError(f"status must be one of: {allowed}") + return normalized + + +def _normalize_optional_status(status: str | None) -> str | None: + normalized = _normalize_optional_text(status) + if normalized is None: + return None + lowered = normalized.lower() + if lowered not in ALL_STATUSES: + allowed = ", ".join(sorted(ALL_STATUSES)) + raise InvalidInputError(f"status must be one of: {allowed}") + return lowered + + +def _normalize_json_object(value: Any, *, field_name: str, redact: bool) -> dict[str, Any]: + if not isinstance(value, dict): + raise InvalidInputError(f"{field_name} must be a JSON object") + if len(value) > MAX_TOP_LEVEL_KEYS: + raise InvalidInputError(f"{field_name} has too many keys") + normalized = _normalize_json_value(value, field_name=field_name, redact=redact) + if not isinstance(normalized, dict): + raise InvalidInputError(f"{field_name} must be a JSON object") + return normalized + + +def _normalize_json_value(value: Any, *, field_name: str, redact: bool, depth: int = 0) -> Any: + if depth > MAX_JSON_DEPTH: + raise InvalidInputError(f"{field_name} exceeds max depth") + if value is None or isinstance(value, bool | int): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise InvalidInputError(f"{field_name} must not contain non-finite floats") + return value + if isinstance(value, str): + if len(value.encode("utf-8")) > MAX_STRING_BYTES: + raise InvalidInputError(f"{field_name} string is too large") + return _redact_text(value, field_name) if redact else value + if isinstance(value, list): + return [ + _normalize_json_value(item, field_name=f"{field_name}[]", redact=redact, depth=depth + 1) + for item in value + ] + if isinstance(value, dict): + if len(value) > MAX_TOP_LEVEL_KEYS: + raise InvalidInputError(f"{field_name} has too many keys") + normalized: dict[str, Any] = {} + for key, item in value.items(): + if not isinstance(key, str) or not key.strip(): + raise InvalidInputError(f"{field_name} keys must be non-empty strings") + if key in _RAW_OUTPUT_KEYS: + raise InvalidInputError(f"{field_name} must not include raw output key {key!r}") + normalized[key] = _normalize_json_value( + item, + field_name=f"{field_name}.{key}", + redact=redact, + depth=depth + 1, + ) + return normalized + raise InvalidInputError(f"{field_name} must contain only JSON-compatible values") + + +def _reject_raw_output_keys(value: Any, field_name: str) -> None: + if isinstance(value, dict): + for key, item in value.items(): + if key in _RAW_OUTPUT_KEYS: + raise InvalidInputError(f"{field_name} must not include raw output key {key!r}") + _reject_raw_output_keys(item, f"{field_name}.{key}") + elif isinstance(value, list): + for index, item in enumerate(value): + _reject_raw_output_keys(item, f"{field_name}[{index}]") + + +def _normalize_digest(value: Any, field_name: str) -> str: + if not isinstance(value, str): + raise InvalidInputError(f"{field_name} must be a SHA-256 hex string") + normalized = value.strip() + if _DIGEST_RE.fullmatch(normalized) is None: + raise InvalidInputError(f"{field_name} must be raw lowercase SHA-256 hex") + return normalized + + +def _normalize_tool_name(value: Any) -> str: + if not isinstance(value, str): + raise InvalidInputError("tool must be a string") + normalized = _require_non_empty(value, "tool") + if _TOOL_NAME_RE.fullmatch(normalized) is None: + raise InvalidInputError("tool must be a normalized tool name") + return normalized + + +def _normalize_positive_int(value: Any, field_name: str) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise InvalidInputError(f"{field_name} must be an integer") + if value <= 0: + raise InvalidInputError(f"{field_name} must be greater than 0") + return int(value) + + +def _normalize_non_negative_int(value: Any, field_name: str) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise InvalidInputError(f"{field_name} must be an integer") + if value < 0: + raise InvalidInputError(f"{field_name} must be non-negative") + return int(value) + + +def _normalize_optional_non_negative_int(value: Any, field_name: str) -> int | None: + if value is None: + return None + return _normalize_non_negative_int(value, field_name) + + +def _normalize_optional_non_negative_float(value: Any, field_name: str) -> float | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int | float) or not math.isfinite(float(value)): + raise InvalidInputError(f"{field_name} must be a finite number") + if float(value) < 0: + raise InvalidInputError(f"{field_name} must be non-negative") + return float(value) + + +def _normalize_optional_redacted_text(value: str | None, field_name: str) -> str | None: + normalized = _normalize_optional_text(value) + if normalized is None: + return None + return _redact_text(normalized, field_name) + + +def _normalize_optional_text(value: str | None) -> str | None: + if value is None: + return None + if not isinstance(value, str): + raise InvalidInputError("value must be a string") + normalized = value.strip() + return normalized or None + + +def _require_non_empty(value: str, field_name: str) -> str: + if not isinstance(value, str): + raise InvalidInputError(f"{field_name} must be a string") + normalized = value.strip() + if not normalized: + raise InvalidInputError(f"{field_name} must not be empty") + return normalized + + +def _redact_text(value: str, field_name: str) -> str: + verdict = redact_secrets(value) + if verdict.verdict is SecretVerdictKind.BLOCK: + raise InvalidInputError(f"{field_name} contains a blocked secret") + return verdict.text + + +def _require_keys(item: dict[str, Any], allowed_keys: set[str], field_name: str) -> None: + keys = set(item) + if keys != allowed_keys: + raise InvalidInputError(f"{field_name} must contain exactly: {', '.join(sorted(allowed_keys))}") + + +def _resolve_history_cutoff(since: str | None, days: int) -> str: + normalized_since = _normalize_optional_text(since) + if normalized_since is not None: + try: + parsed = datetime.fromisoformat(normalized_since.replace("Z", "+00:00")) + except ValueError as exc: + raise InvalidInputError("since must be a valid ISO8601 timestamp") from exc + if parsed.tzinfo is None: + raise InvalidInputError("since must include timezone information") + return parsed.astimezone(UTC).isoformat(timespec="microseconds").replace("+00:00", "Z") + if days <= 0: + raise InvalidInputError("days must be greater than 0") + cutoff = datetime.now(UTC) - timedelta(days=days) + return cutoff.isoformat(timespec="microseconds").replace("+00:00", "Z") + + +def _normalize_limit(limit: int) -> int: + if isinstance(limit, bool) or not isinstance(limit, int): + raise InvalidInputError("limit must be an integer") + if limit <= 0: + raise InvalidInputError("limit must be greater than 0") + return min(limit, MAX_HISTORY_LIMIT) + + +def _json_loads_dict(value: object, field_name: str) -> dict[str, Any]: + if value is None: + return {} + try: + parsed = json.loads(str(value)) + except json.JSONDecodeError as exc: + raise InvalidInputError(f"{field_name} must be valid JSON") from exc + return parsed if isinstance(parsed, dict) else {} + + +def _json_loads_list(value: object, field_name: str) -> list[Any]: + if value is None: + return [] + try: + parsed = json.loads(str(value)) + except json.JSONDecodeError as exc: + raise InvalidInputError(f"{field_name} must be valid JSON") from exc + return parsed if isinstance(parsed, list) else [] + + +def _dump_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) diff --git a/minx_mcp/core/server.py b/minx_mcp/core/server.py index b457954..c4b4d82 100644 --- a/minx_mcp/core/server.py +++ b/minx_mcp/core/server.py @@ -17,6 +17,7 @@ from minx_mcp.core.tools._shared import CoreServiceConfig from minx_mcp.core.tools.enrichment import register_enrichment_tools from minx_mcp.core.tools.goals import register_goal_tools +from minx_mcp.core.tools.investigations import register_investigation_tools from minx_mcp.core.tools.memory import register_memory_tools from minx_mcp.core.tools.playbooks import register_playbook_tools from minx_mcp.core.tools.snapshot import register_snapshot_tools @@ -33,4 +34,5 @@ def create_core_server(config: CoreServiceConfig) -> FastMCP: register_memory_tools(mcp, config) register_enrichment_tools(mcp, config) register_playbook_tools(mcp, config) + register_investigation_tools(mcp, config) return mcp diff --git a/minx_mcp/core/tools/investigations.py b/minx_mcp/core/tools/investigations.py new file mode 100644 index 0000000..d88e549 --- /dev/null +++ b/minx_mcp/core/tools/investigations.py @@ -0,0 +1,278 @@ +"""Investigation lifecycle MCP tools and resources.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from minx_mcp.contracts import InvalidInputError, ToolResponse, wrap_tool_call +from minx_mcp.core import investigations as investigation_api +from minx_mcp.core.tools._shared import CoreServiceConfig +from minx_mcp.db import scoped_connection + +__all__ = ["register_investigation_tools"] + + +def register_investigation_tools(mcp: FastMCP, config: CoreServiceConfig) -> None: + @mcp.resource("investigation://recent") + def investigation_recent_resource() -> str: + with scoped_connection(Path(config.db_path)) as conn: + return json.dumps(investigation_api.recent_resource_payload(conn)) + + @mcp.resource("investigation://{investigation_id}") + def investigation_by_id_resource(investigation_id: str) -> str: + try: + parsed_id = int(investigation_id) + except ValueError as exc: + raise InvalidInputError("investigation_id must be an integer") from exc + with scoped_connection(Path(config.db_path)) as conn: + return json.dumps( + investigation_api.investigation_resource_payload( + conn, + investigation_id=parsed_id, + ) + ) + + @mcp.tool(name="start_investigation") + def start_investigation_tool( + kind: str, + question: str, + context_json: dict[str, Any] | None = None, + harness: str = "hermes", + ) -> ToolResponse: + return wrap_tool_call( + lambda: _start_investigation( + config, + kind=kind, + question=question, + context_json=context_json, + harness=harness, + ), + tool_name="start_investigation", + ) + + @mcp.tool(name="append_investigation_step") + def append_investigation_step_tool(investigation_id: int, step_json: dict[str, Any]) -> ToolResponse: + return wrap_tool_call( + lambda: _append_investigation_step(config, investigation_id=investigation_id, step_json=step_json), + tool_name="append_investigation_step", + ) + + @mcp.tool(name="complete_investigation") + def complete_investigation_tool( + investigation_id: int, + status: str, + answer_md: str | None = None, + citation_refs: list[dict[str, Any]] | None = None, + tool_call_count: int | None = None, + token_input: int | None = None, + token_output: int | None = None, + cost_usd: float | None = None, + error_message: str | None = None, + ) -> ToolResponse: + return wrap_tool_call( + lambda: _complete_investigation( + config, + investigation_id=investigation_id, + status=status, + answer_md=answer_md, + citation_refs=citation_refs, + tool_call_count=tool_call_count, + token_input=token_input, + token_output=token_output, + cost_usd=cost_usd, + error_message=error_message, + ), + tool_name="complete_investigation", + ) + + @mcp.tool(name="log_investigation") + def log_investigation_tool( + kind: str, + question: str, + context_json: dict[str, Any] | None = None, + harness: str = "hermes", + trajectory_json: list[dict[str, Any]] | None = None, + status: str = "succeeded", + answer_md: str | None = None, + citation_refs: list[dict[str, Any]] | None = None, + tool_call_count: int | None = None, + token_input: int | None = None, + token_output: int | None = None, + cost_usd: float | None = None, + error_message: str | None = None, + ) -> ToolResponse: + return wrap_tool_call( + lambda: _log_investigation( + config, + kind=kind, + question=question, + context_json=context_json, + harness=harness, + trajectory_json=trajectory_json, + status=status, + answer_md=answer_md, + citation_refs=citation_refs, + tool_call_count=tool_call_count, + token_input=token_input, + token_output=token_output, + cost_usd=cost_usd, + error_message=error_message, + ), + tool_name="log_investigation", + ) + + @mcp.tool(name="investigation_history") + def investigation_history_tool( + kind: str | None = None, + harness: str | None = None, + status: str | None = None, + since: str | None = None, + days: int = 30, + limit: int = 100, + ) -> ToolResponse: + return wrap_tool_call( + lambda: _investigation_history( + config, + kind=kind, + harness=harness, + status=status, + since=since, + days=days, + limit=limit, + ), + tool_name="investigation_history", + ) + + @mcp.tool(name="investigation_get") + def investigation_get_tool(investigation_id: int) -> ToolResponse: + return wrap_tool_call( + lambda: _investigation_get(config, investigation_id=investigation_id), + tool_name="investigation_get", + ) + + +def _start_investigation( + config: CoreServiceConfig, + *, + kind: str, + question: str, + context_json: dict[str, Any] | None, + harness: str, +) -> dict[str, object]: + with scoped_connection(Path(config.db_path)) as conn: + return investigation_api.start_investigation( + conn, + kind=kind, + question=question, + context_json=context_json, + harness=harness, + ) + + +def _append_investigation_step( + config: CoreServiceConfig, + *, + investigation_id: int, + step_json: dict[str, Any], +) -> dict[str, object]: + with scoped_connection(Path(config.db_path)) as conn: + return investigation_api.append_investigation_step( + conn, + investigation_id=investigation_id, + step_json=step_json, + ) + + +def _complete_investigation( + config: CoreServiceConfig, + *, + investigation_id: int, + status: str, + answer_md: str | None, + citation_refs: list[dict[str, Any]] | None, + tool_call_count: int | None, + token_input: int | None, + token_output: int | None, + cost_usd: float | None, + error_message: str | None, +) -> dict[str, object]: + with scoped_connection(Path(config.db_path)) as conn: + return investigation_api.complete_investigation( + conn, + investigation_id=investigation_id, + status=status, + answer_md=answer_md, + citation_refs=citation_refs, + tool_call_count=tool_call_count, + token_input=token_input, + token_output=token_output, + cost_usd=cost_usd, + error_message=error_message, + ) + + +def _log_investigation( + config: CoreServiceConfig, + *, + kind: str, + question: str, + context_json: dict[str, Any] | None, + harness: str, + trajectory_json: list[dict[str, Any]] | None, + status: str, + answer_md: str | None, + citation_refs: list[dict[str, Any]] | None, + tool_call_count: int | None, + token_input: int | None, + token_output: int | None, + cost_usd: float | None, + error_message: str | None, +) -> dict[str, object]: + with scoped_connection(Path(config.db_path)) as conn: + return investigation_api.log_investigation( + conn, + kind=kind, + question=question, + context_json=context_json, + harness=harness, + trajectory_json=trajectory_json, + status=status, + answer_md=answer_md, + citation_refs=citation_refs, + tool_call_count=tool_call_count, + token_input=token_input, + token_output=token_output, + cost_usd=cost_usd, + error_message=error_message, + ) + + +def _investigation_history( + config: CoreServiceConfig, + *, + kind: str | None, + harness: str | None, + status: str | None, + since: str | None, + days: int, + limit: int, +) -> dict[str, object]: + with scoped_connection(Path(config.db_path)) as conn: + return investigation_api.investigation_history( + conn, + kind=kind, + harness=harness, + status=status, + since=since, + days=days, + limit=limit, + ) + + +def _investigation_get(config: CoreServiceConfig, *, investigation_id: int) -> dict[str, object]: + with scoped_connection(Path(config.db_path)) as conn: + return investigation_api.investigation_get(conn, investigation_id=investigation_id) diff --git a/minx_mcp/schema/migrations/027_investigations.sql b/minx_mcp/schema/migrations/027_investigations.sql new file mode 100644 index 0000000..ebb3949 --- /dev/null +++ b/minx_mcp/schema/migrations/027_investigations.sql @@ -0,0 +1,36 @@ +-- Slice 9: durable investigation lifecycle and render-event audit trail. +CREATE TABLE investigations ( + id INTEGER PRIMARY KEY, + harness TEXT NOT NULL, + kind TEXT NOT NULL + CHECK (kind IN ('investigate', 'plan', 'retro', 'onboard', 'other')), + question TEXT NOT NULL, + context_json TEXT, + status TEXT NOT NULL DEFAULT 'running' + CHECK (status IN ('running', 'succeeded', 'failed', 'cancelled', 'budget_exhausted')), + answer_md TEXT, + trajectory_json TEXT NOT NULL DEFAULT '[]', + response_template TEXT, + response_slots_json TEXT, + citation_refs_json TEXT, + tool_call_count INTEGER, + token_input INTEGER, + token_output INTEGER, + cost_usd REAL, + started_at TEXT NOT NULL, + completed_at TEXT, + error_message TEXT +); + +CREATE INDEX idx_investigations_kind_started +ON investigations(kind, started_at DESC); + +CREATE INDEX idx_investigations_harness_started +ON investigations(harness, started_at DESC); + +CREATE INDEX idx_investigations_status_started +ON investigations(status, started_at DESC); + +CREATE INDEX idx_investigations_running +ON investigations(status) +WHERE status = 'running'; diff --git a/tests/test_core_server.py b/tests/test_core_server.py index fade945..4f781df 100644 --- a/tests/test_core_server.py +++ b/tests/test_core_server.py @@ -28,6 +28,12 @@ def test_core_server_registers_slice25_tool_names(tmp_path: Path) -> None: "memory_reject", "memory_expire", "get_pending_memory_candidates", + "start_investigation", + "append_investigation_step", + "complete_investigation", + "log_investigation", + "investigation_history", + "investigation_get", "list_snapshot_archives", "get_snapshot_archive", } diff --git a/tests/test_investigations.py b/tests/test_investigations.py new file mode 100644 index 0000000..3aa3b5b --- /dev/null +++ b/tests/test_investigations.py @@ -0,0 +1,272 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from minx_mcp.contracts import CONFLICT, INVALID_INPUT +from minx_mcp.core.investigations import canonical_json_digest +from minx_mcp.core.server import create_core_server +from minx_mcp.db import get_connection +from tests.helpers import MinxTestConfig, call_tool_sync, get_tool + +ARG_DIGEST = "a" * 64 +RESULT_DIGEST = "b" * 64 +SECRET_VALUE = "AKIA" + "A" * 16 + + +def _server(tmp_path: Path): + return create_core_server(MinxTestConfig(tmp_path / "minx.db", tmp_path / "vault")) + + +def _step( + *, + step: int = 1, + event_template: str = "investigation.step_logged", + event_slots: dict[str, object] | None = None, + tool: str = "finance_query", + args_digest: str = ARG_DIGEST, + result_digest: str = RESULT_DIGEST, + latency_ms: int = 12, +) -> dict[str, object]: + return { + "step": step, + "event_template": event_template, + "event_slots": event_slots or {"row_count": 3}, + "tool": tool, + "args_digest": args_digest, + "result_digest": result_digest, + "latency_ms": latency_ms, + } + + +def test_canonical_json_digest_is_order_independent_raw_sha256() -> None: + left = canonical_json_digest({"b": 2, "a": [1, {"c": "x"}]}) + right = canonical_json_digest({"a": [1, {"c": "x"}], "b": 2}) + + assert left == right + assert len(left) == 64 + assert left == left.lower() + assert not left.startswith("sha256:") + + +def test_investigation_lifecycle_stores_render_citations_and_filters_history(tmp_path: Path) -> None: + server = _server(tmp_path) + start = get_tool(server, "start_investigation").fn + append = get_tool(server, "append_investigation_step").fn + complete = get_tool(server, "complete_investigation").fn + history = get_tool(server, "investigation_history").fn + get = get_tool(server, "investigation_get").fn + + created = call_tool_sync(start, "investigate", "Why did dining increase?", {"period": "2026-04"}, "hermes") + + assert created["success"] is True + investigation_id = created["data"]["investigation_id"] + assert created["data"]["response_template"] == "investigation.started" + assert created["data"]["response_slots"]["status"] == "running" + + appended = call_tool_sync(append, investigation_id, _step()) + + assert appended["success"] is True + assert appended["data"]["response_template"] == "investigation.step_logged" + assert appended["data"]["response_slots"]["tool"] == "finance_query" + + citations = [ + {"type": "memory", "id": 123}, + {"type": "tool_result_digest", "tool": "finance_query", "digest": RESULT_DIGEST}, + ] + completed = call_tool_sync( + complete, + investigation_id, + "succeeded", + "Dining increased because restaurant spend rose.", + citations, + 1, + 100, + 50, + 0.01, + None, + ) + + assert completed["success"] is True + assert completed["data"]["response_template"] == "investigation.completed" + assert completed["data"]["response_slots"]["status"] == "succeeded" + assert completed["data"]["response_slots"]["citation_count"] == 2 + assert completed["data"]["response_slots"]["cited_memory_count"] == 1 + + run = call_tool_sync(get, investigation_id)["data"]["run"] + assert run["answer_md"] == "Dining increased because restaurant spend rose." + assert run["citation_refs"] == citations + assert run["trajectory"][0]["result_digest"] == RESULT_DIGEST + assert run["response_template"] == "investigation.completed" + + filtered = call_tool_sync(history, "investigate", "hermes", "succeeded", None, 30, 10)["data"] + assert filtered["truncated"] is False + assert [item["investigation_id"] for item in filtered["runs"]] == [investigation_id] + + empty = call_tool_sync(history, "investigate", "other-harness", "succeeded", None, 30, 10)["data"] + assert empty["runs"] == [] + + +def test_append_rejects_bad_digest_raw_output_and_terminal_runs(tmp_path: Path) -> None: + server = _server(tmp_path) + start = get_tool(server, "start_investigation").fn + append = get_tool(server, "append_investigation_step").fn + complete = get_tool(server, "complete_investigation").fn + investigation_id = call_tool_sync(start, "investigate", "Probe", {}, "hermes")["data"]["investigation_id"] + + bad_digest = call_tool_sync(append, investigation_id, _step(args_digest="sha256:" + ARG_DIGEST)) + assert bad_digest["success"] is False + assert bad_digest["error_code"] == INVALID_INPUT + + raw_output = call_tool_sync(append, investigation_id, _step(event_slots={"raw_output": "full rows"})) + assert raw_output["success"] is False + assert raw_output["error_code"] == INVALID_INPUT + + assert call_tool_sync(append, investigation_id, _step())["success"] is True + assert call_tool_sync(complete, investigation_id, "failed", None, [], 1, 0, 0, None, "stopped")["success"] is True + + after_terminal = call_tool_sync(append, investigation_id, _step(step=2)) + assert after_terminal["success"] is False + assert after_terminal["error_code"] == CONFLICT + + +def test_append_event_slots_cannot_override_lifecycle_response_slots(tmp_path: Path) -> None: + server = _server(tmp_path) + start = get_tool(server, "start_investigation").fn + append = get_tool(server, "append_investigation_step").fn + investigation_id = call_tool_sync(start, "investigate", "Probe", {}, "hermes")["data"]["investigation_id"] + + result = call_tool_sync( + append, + investigation_id, + _step( + event_slots={ + "investigation_id": 999, + "status": "succeeded", + "kind": "retro", + "harness": "other-harness", + "tool": "spoofed_tool", + "action": "kept", + } + ), + ) + + slots = result["data"]["response_slots"] + assert slots["investigation_id"] == investigation_id + assert slots["status"] == "running" + assert slots["kind"] == "investigate" + assert slots["harness"] == "hermes" + assert slots["tool"] == "finance_query" + assert slots["action"] == "kept" + + +def test_confirmation_step_returns_confirmation_without_terminal_status(tmp_path: Path) -> None: + server = _server(tmp_path) + start = get_tool(server, "start_investigation").fn + append = get_tool(server, "append_investigation_step").fn + get = get_tool(server, "investigation_get").fn + investigation_id = call_tool_sync(start, "investigate", "Should I promote this?", {}, "hermes")["data"][ + "investigation_id" + ] + + result = call_tool_sync( + append, + investigation_id, + _step( + event_template="investigation.needs_confirmation", + event_slots={"action": "memory_confirm", "risk": "promote_memory", "target_id": 123}, + ), + ) + + assert result["success"] is True + assert result["data"]["response_template"] == "investigation.needs_confirmation" + assert result["data"]["response_slots"]["action"] == "memory_confirm" + assert call_tool_sync(get, investigation_id)["data"]["run"]["status"] == "running" + + +def test_redacts_persisted_text_and_blocks_non_redactable_secrets(tmp_path: Path) -> None: + server = _server(tmp_path) + start = get_tool(server, "start_investigation").fn + complete = get_tool(server, "complete_investigation").fn + + created = call_tool_sync(start, "investigate", f"Review {SECRET_VALUE}", {"note": SECRET_VALUE}, "hermes") + investigation_id = created["data"]["investigation_id"] + completed = call_tool_sync( + complete, + investigation_id, + "succeeded", + f"Do not expose {SECRET_VALUE}", + [{"type": "vault_path", "path": f"Notes/{SECRET_VALUE}.md"}], + 0, + 0, + 0, + None, + None, + ) + + assert completed["success"] is True + conn = get_connection(tmp_path / "minx.db") + row = conn.execute("SELECT question, context_json, answer_md, citation_refs_json FROM investigations").fetchone() + assert SECRET_VALUE not in row["question"] + assert SECRET_VALUE not in row["context_json"] + assert SECRET_VALUE not in row["answer_md"] + assert SECRET_VALUE not in row["citation_refs_json"] + assert "[REDACTED:aws_access_key_id]" in row["question"] + + private_key = "-----BEGIN PRIVATE KEY-----\nabc\n-----END PRIVATE KEY-----" + blocked = call_tool_sync(start, "investigate", private_key, {}, "hermes") + assert blocked["success"] is False + assert blocked["error_code"] == INVALID_INPUT + + +def test_log_investigation_convenience_wrapper_persists_terminal_row(tmp_path: Path) -> None: + server = _server(tmp_path) + log = get_tool(server, "log_investigation").fn + + result = call_tool_sync( + log, + "retro", + "What changed this week?", + {"period": "week"}, + "hermes", + [_step()], + "budget_exhausted", + "Partial answer.", + [{"type": "investigation", "id": 7}], + 1, + 10, + 20, + None, + None, + ) + + assert result["success"] is True + assert result["data"]["response_template"] == "investigation.budget_exhausted" + run = call_tool_sync(get_tool(server, "investigation_get").fn, result["data"]["investigation_id"])["data"]["run"] + assert run["status"] == "budget_exhausted" + assert run["trajectory"][0]["step"] == 1 + assert run["citation_refs"] == [{"type": "investigation", "id": 7}] + + +@pytest.mark.asyncio +async def test_investigation_resources_expose_recent_and_by_id(tmp_path: Path) -> None: + server = _server(tmp_path) + start = get_tool(server, "start_investigation").fn + investigation_id = call_tool_sync(start, "investigate", "Resource probe", {}, "hermes")["data"][ + "investigation_id" + ] + + resource_uris = {str(resource.uri) for resource in await server.list_resources()} + template_uris = {template.uriTemplate for template in await server.list_resource_templates()} + assert "investigation://recent" in resource_uris + assert "investigation://{investigation_id}" in template_uris + + recent_contents = await server.read_resource("investigation://recent") + recent = json.loads(next(iter(recent_contents)).content) + assert recent["runs"][0]["investigation_id"] == investigation_id + + run_contents = await server.read_resource(f"investigation://{investigation_id}") + run = json.loads(next(iter(run_contents)).content)["run"] + assert run["investigation_id"] == investigation_id diff --git a/tests/test_memory_service.py b/tests/test_memory_service.py index e610dc7..037d2bd 100644 --- a/tests/test_memory_service.py +++ b/tests/test_memory_service.py @@ -702,7 +702,8 @@ def test_migration_set_includes_015_memories_unique_live() -> None: assert "024_memory_embeddings.sql" in names assert "025_memory_fts_aliases.sql" in names assert "026_memory_capture_fts.sql" in names - assert names[-1] == "026_memory_capture_fts.sql" + assert "027_investigations.sql" in names + assert names[-1] == "027_investigations.sql" def test_unique_index_rejects_duplicate_live_triple(tmp_path) -> None: