From 00572b57d7bbfe337cff6676f70b76041a7a533d Mon Sep 17 00:00:00 2001 From: Raphael Southall <69826455+raphasouthall@users.noreply.github.com> Date: Tue, 25 Aug 2026 15:07:42 +0100 Subject: [PATCH] feat(harvest): harvest a posted transcript over MCP, add omp provider Harvest could only see session files on the machine running the server, so capture depended on rsyncing transcripts to it first. Extract the classify/redact/dedup/save loop out of harvest_sessions into _harvest_messages, and add harvest_transcript on top of it: the client posts its own session text, the named provider parses it, and the rest of the path is unchanged. Redaction sits inside the shared seam, so the issue #113 contract holds for both entry points by construction. Chunking stays client-side. JSONL is line-oriented, so a chunk split on a newline yields whole messages, each chunk is harvested independently, and the 0.88 cosine dedup absorbs any overlap. A re-post guard keyed mcp:: makes an identical retry a no-op. Also add OmpProvider for ~/.omp/agent/sessions/*/*.jsonl so the file-scan path and backfills understand omp sessions. Its toolResult messages carry raw tool output and are excluded, as are thinking and toolCall parts. Part of #115 --- README.md | 1 + src/neurostack/harvest.py | 364 ++++++++++++++++++++------ src/neurostack/tools/session_tools.py | 38 +++ tests/test_harvest.py | 231 ++++++++++++++++ 4 files changed, 552 insertions(+), 82 deletions(-) diff --git a/README.md b/README.md index 9423322..fc3c633 100644 --- a/README.md +++ b/README.md @@ -322,6 +322,7 @@ include_observations = false # also write the noisier observation/context type | `vault_forget` | Delete a memory | | `vault_memories` | List or search memories | | `vault_harvest` | Extract insights from session transcripts | +| `vault_harvest_transcript` | Extract insights from a transcript posted by the client (no server filesystem access) | **Sessions** diff --git a/src/neurostack/harvest.py b/src/neurostack/harvest.py index 622d35a..5511d26 100644 --- a/src/neurostack/harvest.py +++ b/src/neurostack/harvest.py @@ -36,8 +36,8 @@ def _harvest_state_path() -> Path: return get_config().db_dir / "harvest_state.json" -def _load_harvest_state() -> dict[str, float]: - """Load harvest state: mapping of session file path -> mtime at harvest.""" +def _load_harvest_state() -> dict[str, float | str]: + """Load harvest state: session path -> mtime, or ``mcp:`` key -> transcript hash.""" path = _harvest_state_path() if path.exists(): try: @@ -47,7 +47,7 @@ def _load_harvest_state() -> dict[str, float]: return {} -def _save_harvest_state(state: dict[str, float]) -> None: +def _save_harvest_state(state: dict[str, float | str]) -> None: """Persist harvest state atomically (temp file + os.replace).""" import os import tempfile @@ -380,6 +380,60 @@ def _extract_gemini_content(content) -> str | None: return None +class OmpProvider: + """Oh My Pi — ~/.omp/agent/sessions/*/*.jsonl + + One JSONL per session, under a per-project subdirectory. Message lines are + {"type": "message", "message": {"role": ..., "content": [part, ...]}}. + Roles also include "toolResult", whose text parts are raw tool output — + excluded, or the pre-filter drowns in it. + """ + + name = "omp" + + def find_sessions(self, n: int) -> list[SessionFile]: + sessions_dir = Path.home() / ".omp" / "agent" / "sessions" + if not sessions_dir.exists(): + return [] + sessions = [] + for f in sessions_dir.glob("*/*.jsonl"): + try: + st = f.stat() + sessions.append(SessionFile(path=f, mtime=st.st_mtime, provider=self.name)) + except OSError: + continue + sessions.sort(key=lambda s: s.mtime, reverse=True) + return sessions[:n] + + def extract_messages(self, path: Path) -> list[Message]: + messages = [] + for entry in _parse_jsonl(path): + if entry.get("type") != "message": + continue + msg = entry.get("message") + if not isinstance(msg, dict): + continue + role = msg.get("role", "") + if role not in ("assistant", "user"): + continue + content = msg.get("content") + if isinstance(content, str): + text = content + elif isinstance(content, list): + # Only "text" parts — "thinking" and "toolCall" parts are noise. + parts = [ + p["text"] for p in content + if isinstance(p, dict) and p.get("type") == "text" + and isinstance(p.get("text"), str) + ] + text = "\n".join(parts) + else: + continue + if text: + messages.append(Message(role=role, text=text)) + return messages + + # Provider registry — order doesn't matter, all are scanned _PROVIDERS: list[SessionProvider] = [ ClaudeCodeProvider(), @@ -387,6 +441,7 @@ def _extract_gemini_content(content) -> str | None: CodexCLIProvider(), AiderProvider(), GeminiCLIProvider(), + OmpProvider(), ] _PROVIDER_MAP: dict[str, SessionProvider] = {p.name: p for p in _PROVIDERS} @@ -698,7 +753,113 @@ def _llm_classify( # --------------------------------------------------------------------------- -# Main harvest entry point +# Shared harvest core +# --------------------------------------------------------------------------- + +def _harvest_messages( + conn, + messages: list[Message], + provider: str, + *, + cfg, + embed_url: str | None, + dry_run: bool, + use_llm: bool, + saved: list[dict], + skipped: list[dict], + counts: dict[str, int], +) -> None: + """Classify one transcript's messages and save the keepers. + + The seam shared by both entry points: ``harvest_sessions`` (session files on + this machine's disk) and ``harvest_transcript`` (a transcript posted over + MCP). Results accumulate into the caller's ``saved``/``skipped``/``counts``, + so a multi-session caller needs no per-session merge step. Redaction lives + here, ahead of the dedup check, so both callers inherit the issue #113 + contract by construction. + """ + from .memories import save_memory + + candidates = [] + + for msg in messages: + if not msg.text or len(msg.text) < _MIN_LEN: + continue + # Skip user messages that are system XML or very long pastes + if msg.role == "user" and (len(msg.text) > 1000 or msg.text.startswith("<")): + continue + + prefilter_type = _prefilter_classify(msg.text, msg.role) + if not prefilter_type: + continue + + candidates.append({ + "text": msg.text, + "role": msg.role, + "prefilter_type": prefilter_type, + "provider": provider, + }) + + # Tier 2: LLM classification + if use_llm and candidates: + classified = _llm_classify(candidates, cfg.llm_url, cfg.llm_model) + else: + # Fallback: regex classification + naive summary + classified = [] + for c in candidates: + c["entity_type"] = c["prefilter_type"] + c["summary"] = _make_summary(c["text"]) + classified.append(c) + + # Save classified insights + for item in classified: + summary = item.get("summary", _make_summary(item["text"])) + # Transcripts carry live credentials; a summary must never store one + # (issue #113). Redact BEFORE the dedup check so the stored form and + # the deduped form are the same string. + summary, redacted = redact_secrets(summary) + etype = item.get("entity_type", item.get("prefilter_type", "observation")) + + if len(summary) < _MIN_LEN: + continue + + tags = _extract_tags(item["text"]) + # Harvest-created rows only: agent-written memories keep their + # caller-chosen TTL. Auto-captured context goes stale in a week; + # auto-captured observations get 30 days to be synthesized into a + # learning (issue #36) before they expire as noise. + ttl = {"context": 168.0, "observation": 720.0}.get(etype) + record = {"content": summary, "entity_type": etype, "tags": tags, + "ttl_hours": ttl, "provider": provider} + if redacted: + record["redacted"] = redacted + + if _is_duplicate(conn, summary, etype, embed_url=embed_url): + record["status"] = "skipped (duplicate)" + skipped.append(record) + continue + + if dry_run: + record["status"] = "would save" + saved.append(record) + else: + try: + mem = save_memory( + conn, content=summary, tags=tags, entity_type=etype, + source_agent=f"harvest/{provider}", ttl_hours=ttl, + embed_url=embed_url, + ) + record["memory_id"] = mem.memory_id + record["status"] = "saved" + saved.append(record) + except Exception as exc: + record["status"] = f"error: {exc}" + skipped.append(record) + counts[etype] = counts.get(etype, 0) + 1 + + +# --------------------------------------------------------------------------- +# Main harvest entry points # --------------------------------------------------------------------------- def harvest_sessions( @@ -722,7 +883,6 @@ def harvest_sessions( provider: Restrict to a single provider name, or None for all. """ from .config import get_config - from .memories import save_memory from .schema import DB_PATH, get_db cfg = get_config() @@ -738,7 +898,8 @@ def harvest_sessions( sessions = [] for s in all_sessions: prev_mtime = harvest_state.get(str(s.path)) - if prev_mtime is not None and prev_mtime == s.mtime: + # `mcp:` keys hold a transcript hash, never an mtime — compare numbers only. + if isinstance(prev_mtime, (int, float)) and prev_mtime == s.mtime: log.debug("Skipping already-harvested session: %s (%s)", s.path.name, s.provider) continue sessions.append(s) @@ -751,93 +912,132 @@ def harvest_sessions( counts: dict[str, int] = {} for session in sessions: - messages = extract_messages(session) - candidates = [] + _harvest_messages( + conn, extract_messages(session), session.provider, + cfg=cfg, embed_url=url, dry_run=dry_run, use_llm=use_llm, + saved=saved, skipped=skipped, counts=counts, + ) - for msg in messages: - if not msg.text or len(msg.text) < _MIN_LEN: - continue - # Skip user messages that are system XML or very long pastes - if msg.role == "user" and (len(msg.text) > 1000 or msg.text.startswith("<")): - continue + # Record harvested sessions (skip on dry run) + if not dry_run: + for s in sessions: + harvest_state[str(s.path)] = s.mtime + _save_harvest_state(harvest_state) - prefilter_type = _prefilter_classify(msg.text, msg.role) - if not prefilter_type: - continue + return { + "sessions_scanned": len(sessions), + "providers": list({s.provider for s in sessions}), + "counts": counts, + "saved": saved, + "skipped": skipped, + "dry_run": dry_run, + } - candidates.append({ - "text": msg.text, - "role": msg.role, - "prefilter_type": prefilter_type, - "provider": session.provider, - }) - # Tier 2: LLM classification - if use_llm and candidates: - classified = _llm_classify(candidates, cfg.llm_url, cfg.llm_model) - else: - # Fallback: regex classification + naive summary - classified = [] - for c in candidates: - c["entity_type"] = c["prefilter_type"] - c["summary"] = _make_summary(c["text"]) - classified.append(c) - - # Save classified insights - for item in classified: - summary = item.get("summary", _make_summary(item["text"])) - # Transcripts carry live credentials; a summary must never store one - # (issue #113). Redact BEFORE the dedup check so the stored form and - # the deduped form are the same string. - summary, redacted = redact_secrets(summary) - etype = item.get("entity_type", item.get("prefilter_type", "observation")) - - if len(summary) < _MIN_LEN: - continue +# Cap on a single posted transcript. Above this the client must split on newline +# boundaries and post each chunk separately: chunks are harvested independently +# and the cosine dedup absorbs whatever the split overlaps. +MAX_TRANSCRIPT_BYTES = 4 * 1024 * 1024 - tags = _extract_tags(item["text"]) - # Harvest-created rows only: agent-written memories keep their - # caller-chosen TTL. Auto-captured context goes stale in a week; - # auto-captured observations get 30 days to be synthesized into a - # learning (issue #36) before they expire as noise. - ttl = {"context": 168.0, "observation": 720.0}.get(etype) - record = {"content": summary, "entity_type": etype, "tags": tags, - "ttl_hours": ttl, "provider": session.provider} - if redacted: - record["redacted"] = redacted - - if _is_duplicate(conn, summary, etype, embed_url=url): - record["status"] = "skipped (duplicate)" - skipped.append(record) - continue - if dry_run: - record["status"] = "would save" - saved.append(record) - else: - try: - mem = save_memory( - conn, content=summary, tags=tags, entity_type=etype, - source_agent=f"harvest/{session.provider}", ttl_hours=ttl, - embed_url=url, - ) - record["memory_id"] = mem.memory_id - record["status"] = "saved" - saved.append(record) - except Exception as exc: - record["status"] = f"error: {exc}" - skipped.append(record) - counts[etype] = counts.get(etype, 0) + 1 +def harvest_transcript( + transcript: str, + session_id: str, + source_agent: str, + dry_run: bool = False, + embed_url: str | None = None, + use_llm: bool = True, +) -> dict: + """Extract insights from a POSTED transcript. Returns report dict. + + The MCP-native counterpart to ``harvest_sessions`` (issue #115): the client + sends its own session text, so the server needs no access to the client's + filesystem. Classification, redaction and dedup are the same code path. + + Args: + transcript: Raw session text in ``source_agent``'s native format. + session_id: Client-side session id, used for the re-post guard. + source_agent: Registered provider name — names the transcript FORMAT. + dry_run: If True, show what would be saved without saving. + embed_url: Override embedding URL. + use_llm: Use LLM for classification (falls back to regex if False). + """ + import hashlib + import tempfile + + from .config import get_config + from .schema import DB_PATH, get_db + + def _err(msg: str) -> dict: + return {"error": msg, "saved": [], "skipped": [], "counts": {}} + + prov = _PROVIDER_MAP.get(source_agent) + if prov is None: + return _err( + f"Unknown source_agent '{source_agent}' — must be one of: " + f"{', '.join(get_provider_names())}" + ) + if not transcript.strip(): + return _err("Empty transcript") + + raw = transcript.encode("utf-8") + if len(raw) > MAX_TRANSCRIPT_BYTES: + return _err( + f"Transcript is {len(raw)} bytes, over the {MAX_TRANSCRIPT_BYTES} " + "byte cap. Split it on newline boundaries and post each chunk as a " + "separate call — chunks are harvested independently and the dedup " + "absorbs any overlap." + ) + + # Re-post guard: same session id + same bytes is a no-op, so a client can + # retry a failed POST without duplicating work. + digest = hashlib.sha256(raw).hexdigest() + state_key = f"mcp:{source_agent}:{session_id}" + harvest_state = _load_harvest_state() + if harvest_state.get(state_key) == digest: + return {"sessions_scanned": 0, "session_id": session_id, + "provider": source_agent, "providers": [source_agent], + "counts": {}, "saved": [], "skipped": [], "dry_run": dry_run, + "note": "transcript already harvested"} + + cfg = get_config() + url = embed_url or cfg.embed_url + conn = get_db(DB_PATH) + + # The provider parsers read a path, so the posted text becomes a temp file + # rather than every provider growing a second entry point. + tmp_path = None + try: + with tempfile.NamedTemporaryFile( + mode="w", suffix=".jsonl", delete=False, encoding="utf-8", + ) as tmp: + tmp_path = tmp.name + tmp.write(transcript) + messages = prov.extract_messages(Path(tmp_path)) + finally: + if tmp_path: + Path(tmp_path).unlink(missing_ok=True) + + saved: list[dict] = [] + skipped: list[dict] = [] + counts: dict[str, int] = {} + if messages: + _harvest_messages( + conn, messages, source_agent, + cfg=cfg, embed_url=url, dry_run=dry_run, use_llm=use_llm, + saved=saved, skipped=skipped, counts=counts, + ) - # Record harvested sessions (skip on dry run) if not dry_run: - for s in sessions: - harvest_state[str(s.path)] = s.mtime + harvest_state[state_key] = digest _save_harvest_state(harvest_state) return { - "sessions_scanned": len(sessions), - "providers": list({s.provider for s in sessions}), + "sessions_scanned": 1, + "session_id": session_id, + "provider": source_agent, + "providers": [source_agent], + "messages": len(messages), "counts": counts, "saved": saved, "skipped": skipped, diff --git a/src/neurostack/tools/session_tools.py b/src/neurostack/tools/session_tools.py index d3c1d8d..a913de1 100644 --- a/src/neurostack/tools/session_tools.py +++ b/src/neurostack/tools/session_tools.py @@ -125,3 +125,41 @@ def vault_harvest(sessions: int = 1, dry_run: bool = False, provider: str | None embed_url=_embed_url(), provider=provider, ) + + +@registry.tool(tags=["session", "memory"], annotations=_WRITE_ADDITIVE) +def vault_harvest_transcript( + transcript: str, + session_id: str, + source_agent: str = "claude-code", + dry_run: bool = False, +) -> dict: + """Extract insights from a session transcript you POST here, and save them. + + Unlike vault_harvest, this reads nothing off the server's filesystem — send + the transcript text and a client on any machine can harvest its own + sessions. Deduplicates against existing memories before saving. + + source_agent names the transcript FORMAT, not the machine: one of + claude-code, vscode-chat, codex-cli, aider, gemini-cli, omp. + + A large transcript must be split on NEWLINE boundaries and posted as + several calls; each chunk is harvested independently and near-duplicate + insights across chunks are dropped. Re-posting an identical transcript for + the same session_id is a no-op. + + Args: + transcript: Raw session transcript in source_agent's native format + session_id: Client-side session identifier (guards against re-posts) + source_agent: Provider name naming the transcript format + dry_run: If True, show what would be saved without saving + """ + from ..harvest import harvest_transcript + + return harvest_transcript( + transcript=transcript, + session_id=session_id, + source_agent=source_agent, + dry_run=dry_run, + embed_url=_embed_url(), + ) diff --git a/tests/test_harvest.py b/tests/test_harvest.py index 3144656..42de194 100644 --- a/tests/test_harvest.py +++ b/tests/test_harvest.py @@ -4,10 +4,12 @@ from types import SimpleNamespace from neurostack.harvest import ( + MAX_TRANSCRIPT_BYTES, AiderProvider, ClaudeCodeProvider, GeminiCLIProvider, Message, + OmpProvider, _extract_gemini_content, _extract_tags, _extract_text_claude, @@ -17,6 +19,8 @@ _parse_jsonl, _prefilter_classify, _save_harvest_state, + get_provider_names, + harvest_transcript, ) # --------------------------------------------------------------------------- @@ -604,3 +608,230 @@ def test_per_type_ttl(self, in_memory_db, tmp_path, monkeypatch): assert rows["context"]["ttl_h"] == 168 assert rows["observation"]["ttl_h"] == 720 assert rows["decision"]["expires_at"] is None + + +# --------------------------------------------------------------------------- +# OmpProvider +# --------------------------------------------------------------------------- + +class TestOmpProvider: + """Message lines carry typed content parts; only "text" parts on the + user/assistant roles are transcript. "toolResult" text is raw tool output.""" + + def test_extract_messages(self, tmp_path): + f = tmp_path / "session.jsonl" + lines = [ + json.dumps({"type": "session", "cwd": "/tmp/proj", "title": "t", + "version": "1"}), + json.dumps({"type": "message", "message": { + "role": "user", + "content": [{"type": "text", "text": "why did the build break"}]}}), + "{not json at all", + json.dumps({"type": "message", "message": { + "role": "assistant", + "content": [ + {"type": "thinking", "text": "let me look"}, + {"type": "text", "text": "the root cause was a stale lockfile"}, + {"type": "text", "text": "the fix was to regenerate it"}, + {"type": "toolCall", "name": "bash"}, + ]}}), + json.dumps({"type": "message", "message": { + "role": "toolResult", + "content": [{"type": "text", "text": "tool output noise"}]}}), + json.dumps({"type": "custom", "customType": "note", "data": {}}), + ] + f.write_text("\n".join(lines) + "\n") + msgs = OmpProvider().extract_messages(f) + assert msgs == [ + Message(role="user", text="why did the build break"), + Message(role="assistant", + text="the root cause was a stale lockfile\n" + "the fix was to regenerate it"), + ] + + def test_string_content_tolerated(self, tmp_path): + f = tmp_path / "session.jsonl" + f.write_text(json.dumps({"type": "message", "message": { + "role": "user", "content": "plain string body"}}) + "\n") + assert OmpProvider().extract_messages(f) == [ + Message(role="user", text="plain string body"), + ] + + def test_missing_content_skipped(self, tmp_path): + f = tmp_path / "session.jsonl" + lines = [ + json.dumps({"type": "message", "message": {"role": "user"}}), + json.dumps({"type": "message", "message": "not a dict"}), + ] + f.write_text("\n".join(lines) + "\n") + assert OmpProvider().extract_messages(f) == [] + + def test_find_sessions(self, tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + proj = tmp_path / ".omp" / "agent" / "sessions" / "-tools-neurostack" + proj.mkdir(parents=True) + f = proj / "2026-01-01T00-00-00Z_abc.jsonl" + f.write_text("{}\n") + found = OmpProvider().find_sessions(5) + assert [s.path for s in found] == [f] + assert found[0].provider == "omp" + + def test_registered(self): + assert "omp" in get_provider_names() + + +# --------------------------------------------------------------------------- +# harvest_transcript — MCP-native harvest (issue #115) +# --------------------------------------------------------------------------- + +# Same shape as a Google API key, none of its characters — assembled here so no +# secret-shaped literal is ever committed. +FAKE_GOOGLE_KEY = "AIza" + "Sy" + "B" * 33 + + +def _claude_line(role, text): + return json.dumps({"message": {"role": role, "content": text}}) + + +_BUG_INSIGHT = ("The root cause was a stale resolver cache, and the fix was to " + "invalidate the entry on every write.") +_DECISION_INSIGHT = ("We decided to keep the harvest dedup threshold at 0.88 " + "rather than tightening it for posted chunks.") + + +class TestHarvestTranscript: + """The client posts its own transcript, so the server needs no access to the + client's filesystem (issue #115).""" + + @staticmethod + def _setup(in_memory_db, tmp_path, monkeypatch): + import zlib + + import numpy as np + + import neurostack.embedder as embedder_mod + import neurostack.harvest as harvest_mod + + monkeypatch.setattr(harvest_mod, "_harvest_state_path", + lambda: tmp_path / "state.json") + monkeypatch.setattr("neurostack.schema.get_db", lambda path: in_memory_db) + cfg = SimpleNamespace(embed_url="http://embed.test", + llm_url="http://llm.test", llm_model="m", + llm_api_key=None, writeback_enabled=False) + monkeypatch.setattr("neurostack.config.get_config", lambda: cfg) + + def embed(content, *a, **k): + # One-hot per distinct text: unrelated insights stay orthogonal, + # a re-posted identical insight still lands on cosine 1.0. + v = np.zeros(768, dtype=np.float32) + v[zlib.crc32(content.encode()) % 768] = 1.0 + return v + + monkeypatch.setattr(embedder_mod, "get_embedding", embed) + + @staticmethod + def _memory_contents(conn): + return [r[0] for r in conn.execute("SELECT content FROM memories")] + + def test_saves_and_cleans_up_temp_file(self, in_memory_db, tmp_path, monkeypatch): + self._setup(in_memory_db, tmp_path, monkeypatch) + seen = [] + real = ClaudeCodeProvider.extract_messages + + def spy(self, path): + seen.append(path) + return real(self, path) + + monkeypatch.setattr(ClaudeCodeProvider, "extract_messages", spy) + + report = harvest_transcript( + _claude_line("assistant", _BUG_INSIGHT) + "\n", + session_id="sess-1", source_agent="claude-code", use_llm=False, + ) + assert [r["status"] for r in report["saved"]] == ["saved"] + assert report["session_id"] == "sess-1" + assert report["provider"] == "claude-code" + assert report["counts"] == {"bug": 1} + assert self._memory_contents(in_memory_db) == [_BUG_INSIGHT] + # The transcript went through a temp file that must not outlive the call. + assert seen and not seen[0].exists() + + def test_unknown_source_agent(self, in_memory_db, tmp_path, monkeypatch): + self._setup(in_memory_db, tmp_path, monkeypatch) + report = harvest_transcript( + _claude_line("assistant", _BUG_INSIGHT), session_id="sess-1", + source_agent="not-a-provider", use_llm=False, + ) + assert "not-a-provider" in report["error"] + for name in get_provider_names(): + assert name in report["error"] + assert report["saved"] == [] and report["counts"] == {} + assert self._memory_contents(in_memory_db) == [] + + def test_empty_transcript(self, in_memory_db, tmp_path, monkeypatch): + self._setup(in_memory_db, tmp_path, monkeypatch) + report = harvest_transcript(" \n\n", session_id="s", source_agent="claude-code") + assert "error" in report + assert self._memory_contents(in_memory_db) == [] + + def test_over_cap_asks_for_newline_chunks(self, in_memory_db, tmp_path, monkeypatch): + self._setup(in_memory_db, tmp_path, monkeypatch) + report = harvest_transcript( + "x" * (MAX_TRANSCRIPT_BYTES + 1), session_id="s", + source_agent="claude-code", use_llm=False, + ) + assert "newline" in report["error"] + assert report["saved"] == [] and report["counts"] == {} + assert self._memory_contents(in_memory_db) == [] + + def test_repost_guard(self, in_memory_db, tmp_path, monkeypatch): + self._setup(in_memory_db, tmp_path, monkeypatch) + transcript = _claude_line("assistant", _BUG_INSIGHT) + "\n" + + first = harvest_transcript(transcript, session_id="sess-1", + source_agent="claude-code", use_llm=False) + assert [r["status"] for r in first["saved"]] == ["saved"] + + again = harvest_transcript(transcript, session_id="sess-1", + source_agent="claude-code", use_llm=False) + assert again["note"] == "transcript already harvested" + assert again["saved"] == [] and again["counts"] == {} + + # A changed transcript for the same session is harvested again: the + # repeated insight dedups, the new one saves. This is what makes + # client-side chunking with overlap safe. + grown = transcript + _claude_line("assistant", _DECISION_INSIGHT) + "\n" + changed = harvest_transcript(grown, session_id="sess-1", + source_agent="claude-code", use_llm=False) + assert "note" not in changed + assert [r["status"] for r in changed["saved"]] == ["saved"] + assert [r["status"] for r in changed["skipped"]] == ["skipped (duplicate)"] + assert sorted(self._memory_contents(in_memory_db)) == sorted( + [_BUG_INSIGHT, _DECISION_INSIGHT] + ) + + def test_redacts_before_storing(self, in_memory_db, tmp_path, monkeypatch): + # The #113 contract holds on the posted path too: the shared seam + # redacts before both the dedup check and the save. + self._setup(in_memory_db, tmp_path, monkeypatch) + text = (f"The root cause was the hardcoded apiKey: {FAKE_GOOGLE_KEY} in " + "the deploy script, which nothing ever rotated.") + report = harvest_transcript( + _claude_line("assistant", text) + "\n", session_id="sess-1", + source_agent="claude-code", use_llm=False, + ) + assert report["saved"][0]["redacted"] == ["google-api-key"] + stored = self._memory_contents(in_memory_db) + assert len(stored) == 1 + assert "***REDACTED***" in stored[0] + assert FAKE_GOOGLE_KEY not in stored[0] + + def test_no_messages_is_not_an_error(self, in_memory_db, tmp_path, monkeypatch): + self._setup(in_memory_db, tmp_path, monkeypatch) + report = harvest_transcript( + json.dumps({"message": {"role": "system", "content": "ignored"}}) + "\n", + session_id="sess-1", source_agent="claude-code", use_llm=False, + ) + assert "error" not in report + assert report["messages"] == 0 + assert report["counts"] == {} and report["saved"] == []