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"] == []