diff --git a/CHANGELOG.md b/CHANGELOG.md index 298e1e2..c311779 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,27 @@ Hy Memory and MemHub are versioned and released independently. ## Unreleased +- MemHub host capture hooks no longer write the database. `one-memory + capture ` used to open the main SQLite file from its own short-lived + process on every hook, re-create the retired `evidence` / FTS tables on an + SDK-authority home and write the write queue outside Core's single writer, + contending for the file lock with Core and with other hosts' hooks. It now + reads only the runtime receipt and token and posts the hook's JSON to the + running Core (`POST /api/v1/live-capture/{source}`, 1.5 s budget). Core + applies the same gates (grant, sync switch, rejection counting, automated + sessions, the Cursor/Claude Code rule; the migration fence stays on the hook + side) and stages the turn in-process through the live file-sync path. When + Core is not running the turn is dropped silently; nothing falls back to a + local write. +- MemHub no longer sends one turn to the SDK twice when a host's hook and live + file sync both see it. The two lanes gave the same message different + identities (the hook carries no workspace and often no message id) and + separate assembly rows, so both were written as raws. Hook input for a + source that live sync reads from its own files (every hooked host except + Qwen Work, while live sync is on) is now left to the file path; only a + source without file polling is staged from its hook. Tables an older hook + re-created on an SDK home are dropped by the existing startup cleanup. + ## Hy Memory 2.0.14 — 2026-11-04 - `search()` / `async_search()` take an optional `statuses` for explicit diff --git a/memhub/apps/core/src/one_memory/api.py b/memhub/apps/core/src/one_memory/api.py index 95b933a..dc06b73 100644 --- a/memhub/apps/core/src/one_memory/api.py +++ b/memhub/apps/core/src/one_memory/api.py @@ -2206,6 +2206,19 @@ def create_app( from one_memory.embedding_switch_routes import embedding_switch_router app.include_router(embedding_switch_router(protected)) + from one_memory.live_capture import live_capture_router + + def file_polled(request: Request, source_id: str) -> bool: + """Whether live sync reads this source's own files, so the hook's copy is redundant.""" + registration = getattr(request.app.state, "source_catalog", {}).get(source_id) + return ( + runtime_settings.live_sync_enabled + and registration is not None + and registration.capabilities.live_sync is CapabilityState.IMPLEMENTED + ) + + # Host hooks hand their turn over here instead of opening the database. + app.include_router(live_capture_router(protected, file_polled=file_polled)) app.add_middleware( CORSMiddleware, diff --git a/memhub/apps/core/src/one_memory/capture_handoff.py b/memhub/apps/core/src/one_memory/capture_handoff.py new file mode 100644 index 0000000..9d75804 --- /dev/null +++ b/memhub/apps/core/src/one_memory/capture_handoff.py @@ -0,0 +1,133 @@ +"""The hook side of live capture: hand the turn to the running Core, nothing more. + +A host runs `one-memory capture ` as a short-lived process on every +hook, often several at once. That process used to open the main database +itself and write Evidence and the write queue from outside Core's single +writer. It now never opens the database: it reads the hook's JSON, posts it to +the running Core over loopback with the runtime token, and exits. Core applies +every gate and stages the turn in-process (`live_capture.capture_hook`). + +Anything that goes wrong — Core not running, a stale receipt, a slow answer — +drops the turn silently. There is no local fallback write: file-backed sources +are read again by live sync from their persisted cursor once Core is back. +""" + +import json +import os +from pathlib import Path +from typing import TextIO + +CAPTURE_SOURCES = frozenset( + { + "workbuddy", + "claude-code", + "codex", + "cursor", + "pi", + "qwen-code", + "qwenwork", + "hermes", + "deepseek-harness", + "opencode", + "openclaw", + } +) +SDK_MIGRATION_CAPTURE_FENCE = ".sdk-migration-capture-fence.json" +#: Core's in-process intake for a hook's payload; see `live_capture`. +CAPTURE_ROUTE = "/api/v1/live-capture/{source_id}" +#: Whole request, connect to answer. Pi and OpenCode wait for the hook +#: synchronously (3 s, interpreter start included), so this stays well under. +HANDOFF_TIMEOUT_SECONDS = 1.5 +#: A hook carries one turn, or one session's visible text at most. +MAX_PAYLOAD_BYTES = 8 * 1024 * 1024 + + +def data_directory() -> Path: + from one_memory.config import load_settings + + return load_settings().data_dir + + +def capture_stdio(source_id: str, stdin: TextIO, output: TextIO) -> int: + """Host hooks must stay silent. A bad payload or an absent Core does not fail the turn.""" + + del output + if source_id not in CAPTURE_SOURCES: + return 2 + try: + data_dir = data_directory() + if sdk_migration_capture_fenced(data_dir): + return 0 + payload = json.loads(stdin.read() or "") + except Exception: + return 0 + if not isinstance(payload, dict): + return 0 + try: + hand_to_core(source_id, payload, data_dir) + except Exception: + return 0 + return 0 + + +def hand_to_core( + source_id: str, + payload: dict[str, object], + data_dir: Path, + *, + timeout: float = HANDOFF_TIMEOUT_SECONDS, +) -> bool: + """Post the payload to the Core named by this home's receipt. True when Core took it. + + Only the receipt, the runtime token and a loopback socket are touched; the + main database is never opened from here. + """ + from one_memory.runtime_client import core_endpoint + + endpoint = core_endpoint(data_dir) + if endpoint is None: + return False + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + if len(body) > MAX_PAYLOAD_BYTES: + return False + from one_memory.http_client import create_client + + with create_client(timeout=timeout) as client: + response = client.post( + f"http://{endpoint.host}:{endpoint.port}{CAPTURE_ROUTE.format(source_id=source_id)}", + content=body, + headers={ + "Authorization": f"Bearer {endpoint.token}", + "Content-Type": "application/json", + }, + ) + return response.status_code == 202 + + +def sdk_migration_capture_fenced(data_dir: Path) -> bool: + """Ignore stale fences; a live migration or a failed one pauses hook writes. + + A failed upgrade migration leaves the fence `frozen` (no pid): the home was + put back as it was before the upgrade and takes no writes until the next + migration attempt, which replaces the fence and lifts it on success. + """ + marker = data_dir / SDK_MIGRATION_CAPTURE_FENCE + try: + document = json.loads(marker.read_text(encoding="utf-8")) + if isinstance(document, dict) and document.get("frozen") is True: + return True + pid = int(document["pid"]) + if pid < 1: + raise ValueError("invalid migration fence pid") + os.kill(pid, 0) + return True + except FileNotFoundError: + return False + except ProcessLookupError: + marker.unlink(missing_ok=True) + return False + except PermissionError: + return True + except (KeyError, TypeError, ValueError, json.JSONDecodeError, OSError): + marker.unlink(missing_ok=True) + return False diff --git a/memhub/apps/core/src/one_memory/cli.py b/memhub/apps/core/src/one_memory/cli.py index 354a820..dc67fb1 100644 --- a/memhub/apps/core/src/one_memory/cli.py +++ b/memhub/apps/core/src/one_memory/cli.py @@ -119,7 +119,7 @@ def run(arguments: Sequence[str], *, output: TextIO | None = None) -> int: if command == "sdk-migration-status": return sdk_migration_status(options, stream) if command == "capture" and len(options) == 1: - from one_memory.live_capture import capture_stdio + from one_memory.capture_handoff import capture_stdio return capture_stdio(options[0], sys.stdin, stream) if command not in {"status", "doctor"} or options not in ([], ["--json"]): diff --git a/memhub/apps/core/src/one_memory/live_capture.py b/memhub/apps/core/src/one_memory/live_capture.py index 0853390..d4a7c4d 100644 --- a/memhub/apps/core/src/one_memory/live_capture.py +++ b/memhub/apps/core/src/one_memory/live_capture.py @@ -1,15 +1,35 @@ -"""Write visible hook text into Evidence. Tool output stays out.""" +"""Core's intake for host hooks: visible text only, staged under Core's writer. +The hook process (`capture_handoff`) never opens the database; it posts the +hook's JSON to `POST /api/v1/live-capture/{source}` on the running Core, which +calls `capture_hook` here. Tool output stays out. + +A source that live sync already reads from its own files is left to that +path: the hook and the file name the same message differently (the hook has +no workspace and often no message id), so staging both sent one turn to the +SDK twice. Only a source no file poller follows is staged from the hook. +""" + +import asyncio import json -import os import time +from collections.abc import Callable from datetime import UTC, datetime from enum import StrEnum from pathlib import Path -from typing import TextIO, cast +from typing import cast +from fastapi import APIRouter, HTTPException, Request +from fastapi.params import Depends from pydantic import JsonValue +from one_memory.capture_handoff import ( + CAPTURE_ROUTE, + CAPTURE_SOURCES, + MAX_PAYLOAD_BYTES, + SDK_MIGRATION_CAPTURE_FENCE, + sdk_migration_capture_fenced, +) from one_memory.database_access import main_database from one_memory.derived_echo import PREFACE, strip_injected_memory from one_memory.imports.access import opaque_workspace_id, safe_project_label @@ -19,6 +39,7 @@ from one_memory.imports.models import ( EvidenceRecord, OriginMetadata, WrittenBy, + materialize_write_input, ) from one_memory.imports.sources.claude_code import parse_claude_code_record from one_memory.imports.sources.codex import HARNESS_PREFIXES, parse_codex_record @@ -38,23 +59,21 @@ from one_memory.imports.sources.parsing import ( ) from one_memory.imports.sources.pi import _is_automated_session from one_memory.imports.sources.workbuddy import _is_scheduled_automation, parse_workbuddy_record -from one_memory.imports.storage import EvidenceStore +from one_memory.memory_server.durable_queue import DurableWriteQueue +from one_memory.sdk_data_plane_migration import MemoryAuthority + +__all__ = [ + "CAPTURE_SOURCES", + "SDK_MIGRATION_CAPTURE_FENCE", + "CaptureOutcome", + "CaptureRejection", + "capture_hook", + "capture_rejection", + "live_capture_router", + "record_capture_rejection", + "sdk_migration_capture_fenced", +] -CAPTURE_SOURCES = frozenset( - { - "workbuddy", - "claude-code", - "codex", - "cursor", - "pi", - "qwen-code", - "qwenwork", - "hermes", - "deepseek-harness", - "opencode", - "openclaw", - } -) _UNKNOWN_USER = frozenset({"hermes", "opencode", "openclaw"}) _USER_FIELDS = { "UserPromptSubmit": "prompt", @@ -75,112 +94,130 @@ _FILE_FALLBACK = { } _FALLBACK_EVENTS = frozenset({"Stop", "stop", "SubagentStop", "StopFailure"}) _TEXT_BLOCKS = frozenset({"text"}) -SDK_MIGRATION_CAPTURE_FENCE = ".sdk-migration-capture-fence.json" -def database_path() -> Path: - from one_memory.config import load_settings +class CaptureOutcome(StrEnum): + """What Core did with one hook payload. Never the payload itself.""" - return load_settings().data_dir / "one-memory.sqlite3" + STAGED = "staged" + #: Live sync reads this source's own files; that path owns the turn. + FILE_POLLED = "file_polled" + REJECTED = "rejected" + IGNORED = "ignored" -def capture_stdio(source_id: str, stdin: TextIO, output: TextIO) -> int: - """Host hooks must stay silent. A bad payload does not fail the turn.""" +def capture_hook( + source_id: str, + payload: object, + database: Path, + *, + queue: DurableWriteQueue, + file_polled: Callable[[str], bool] = lambda _source_id: False, +) -> CaptureOutcome: + """Gate one hook payload and stage its visible lines, inside Core. - del output - if source_id not in CAPTURE_SOURCES: - return 2 - database = database_path() - if sdk_migration_capture_fenced(database.parent): - return 0 - try: - payload = json.loads(stdin.read() or "") - except json.JSONDecodeError: - return 0 - try: - capture_hook(source_id, payload, database) - except Exception: - return 0 - return 0 - - -def sdk_migration_capture_fenced(data_dir: Path) -> bool: - """Ignore stale fences; a live migration or a failed one pauses hook writes. - - A failed upgrade migration leaves the fence `frozen` (no pid): the home was - put back as it was before the upgrade and takes no writes until the next - migration attempt, which replaces the fence and lifts it on success. + Runs on SDK authority through Core's single writer: the lines become + write-job input exactly as live file sync stages them + (`materialize_write_input` -> `stage_sdk_input` -> `finalize_sdk_file`, + live priority). No Evidence table is involved. """ - marker = data_dir / SDK_MIGRATION_CAPTURE_FENCE - try: - document = json.loads(marker.read_text(encoding="utf-8")) - if isinstance(document, dict) and document.get("frozen") is True: - return True - pid = int(document["pid"]) - if pid < 1: - raise ValueError("invalid migration fence pid") - os.kill(pid, 0) - return True - except FileNotFoundError: - return False - except ProcessLookupError: - marker.unlink(missing_ok=True) - return False - except PermissionError: - return True - except (KeyError, TypeError, ValueError, json.JSONDecodeError, OSError): - marker.unlink(missing_ok=True) - return False - - -def capture_hook(source_id: str, payload: object, database: Path) -> bool: if source_id not in CAPTURE_SOURCES or not isinstance(payload, dict): - return False + return CaptureOutcome.IGNORED # Cursor also runs ~/.claude/settings.json hooks; its own hook already files the turn. if source_id != "cursor" and "cursor_version" in payload: - return False + return CaptureOutcome.IGNORED if not database.is_file(): - return False + return CaptureOutcome.IGNORED rejection = capture_rejection(database, source_id) if rejection is not None: # Silent, and the reason is counted rather than printed: a hook that # spoke would put this into the host's turn as text. record_capture_rejection(database, source_id, rejection) - return False + return CaptureOutcome.REJECTED if source_id == "pi" and _is_automated_session(first_string(payload.get("cwd"))): - return False - lines = _messages(source_id, payload) - if lines: - conversation_id = _conversation(payload) - store = EvidenceStore(database) - wrote = False - for role, written_by, verbatim, message_id in lines: - target = conversation_id - if target is None: - continue - result = store.put( - EvidenceInput( - source_id=source_id, - conversation_id=target, - message_id=message_id, - source_ordinal=None if message_id else "hook-visible", - role=role, - written_by=written_by, - created_at=datetime.now(UTC), - verbatim=verbatim, - origin_meta={EvidenceOrigin.HOOK: OriginMetadata()}, - ) - ) - _submit_live_memory( - store, result.record, changed=result.was_created or result.was_revised - ) - wrote = True - return wrote - return _capture_transcript(source_id, payload, database) + return CaptureOutcome.IGNORED + if file_polled(source_id): + # The same turn reaches the SDK once, from the file: the two lanes give + # one message different identities, so staging both wrote it twice. + return CaptureOutcome.FILE_POLLED + inputs = _hook_inputs(source_id, payload) or _capture_transcript(source_id, payload, database) + if not inputs: + return CaptureOutcome.IGNORED + observed = datetime.now(UTC) + conversations: dict[str, list[EvidenceRecord]] = {} + for value in inputs: + conversations.setdefault(value.conversation_id, []).append( + materialize_write_input(value, observed_at=observed) + ) + for conversation_id, records in conversations.items(): + # The turn rule decides what is written (`turn_batches`): a turn is + # complete when the next user message arrives, batches are 5 turns or + # 8000 characters, and Core writes the rest after 10 quiet minutes. + queue.assemble_live_records(tuple(records), file_key=f"hook:{conversation_id}") + return CaptureOutcome.STAGED -def _sync_open(database: Path, source_id: str) -> bool: - return capture_rejection(database, source_id) is None +def _hook_inputs(source_id: str, payload: dict[str, object]) -> list[EvidenceInput]: + conversation_id = _conversation(payload) + if conversation_id is None: + return [] + observed = datetime.now(UTC) + return [ + EvidenceInput( + source_id=source_id, + conversation_id=conversation_id, + message_id=message_id, + source_ordinal=None if message_id else "hook-visible", + role=role, + written_by=written_by, + created_at=observed, + verbatim=verbatim, + origin_meta={EvidenceOrigin.HOOK: OriginMetadata()}, + ) + for role, written_by, verbatim, message_id in _messages(source_id, payload) + ] + + +def live_capture_router( + dependencies: list[Depends], + *, + file_polled: Callable[[Request, str], bool], +) -> APIRouter: + """`POST /api/v1/live-capture/{source_id}`: a hook's payload, handed over by the hook. + + 202 whatever the gates decide, so a hook has nothing to print. The answer + names the outcome only, never any of the payload. + """ + router = APIRouter(dependencies=dependencies) + + @router.post(CAPTURE_ROUTE, status_code=202) + async def live_capture(source_id: str, request: Request) -> dict[str, str]: + if source_id not in CAPTURE_SOURCES: + raise HTTPException(status_code=404, detail="capture_source_not_found") + body = await request.body() + if len(body) > MAX_PAYLOAD_BYTES: + raise HTTPException(status_code=413, detail="capture_payload_too_large") + try: + payload = json.loads(body) + except (json.JSONDecodeError, UnicodeError) as error: + raise HTTPException(status_code=422, detail="capture_payload_invalid") from error + state = request.app.state + queue = getattr(state, "memory_write_queue", None) + if getattr(state, "memory_authority", None) is not MemoryAuthority.SDK or queue is None: + # Only an SDK-authority home is served; nothing here may bring the + # retired Evidence tables back for any other. + return {"outcome": CaptureOutcome.IGNORED.value} + outcome = await asyncio.to_thread( + capture_hook, + source_id, + payload, + queue.database_path, + queue=queue, + file_polled=lambda source: file_polled(request, source), + ) + return {"outcome": outcome.value} + + return router class CaptureRejection(StrEnum): @@ -255,22 +292,6 @@ def record_capture_rejection( return -def _submit_live_memory(store: EvidenceStore, record: EvidenceRecord, *, changed: bool) -> None: - """Add the record to its conversation's live assembling row. - - The turn rule decides what is written (`turn_batches`): a turn is - complete when the next user message arrives, batches are 5 turns or - 8000 characters, and Core writes the rest after 10 quiet minutes. - """ - if not changed or not _sync_open(store.database_path, record.source_id): - return - from one_memory.memory_server.durable_queue import DurableWriteQueue - - DurableWriteQueue(store.database_path, evidence=store).assemble_live_records( - (record,), file_key=f"hook:{record.conversation_id}" - ) - - def _conversation(payload: dict[str, object]) -> str | None: return first_string( cast(JsonValue | None, payload.get("session_id")), @@ -470,34 +491,35 @@ def _opencode_messages( return lines -def _capture_transcript(source_id: str, payload: dict[str, object], database: Path) -> bool: +def _capture_transcript( + source_id: str, payload: dict[str, object], database: Path +) -> list[EvidenceInput]: """Stop without in-memory text reads the granted transcript, then the same parser.""" event = payload.get("hook_event_name") or payload.get("hookEventName") parser = _FILE_FALLBACK.get(source_id) transcript = payload.get("transcript_path") if event not in _FALLBACK_EVENTS or parser is None or not isinstance(transcript, str): - return False + return [] root = _grant_root(database, source_id) if root is None: - return False + return [] try: granted = root.resolve() target = Path(transcript).resolve() except OSError: - return False + return [] if target != granted and granted not in target.parents: - return False + return [] if not target.is_file(): - return False + return [] _wait_stable(target) try: raw = target.read_text(encoding="utf-8") except OSError: - return False + return [] relative = target.relative_to(granted).as_posix() - store = EvidenceStore(database) - wrote = False + inputs: list[EvidenceInput] = [] for index, line in enumerate(raw.splitlines()): if not line.strip(): continue @@ -514,7 +536,7 @@ def _capture_transcript(source_id: str, payload: dict[str, object], database: Pa if not isinstance(parsed, ParsedMessage): continue for row_index, row in enumerate(parsed.rows()): - result = store.put( + inputs.append( EvidenceInput( source_id=source_id, workspace=opaque_workspace_id(source_id, row.workspace), @@ -533,11 +555,7 @@ def _capture_transcript(source_id: str, payload: dict[str, object], database: Pa origin_meta={EvidenceOrigin.FILE: OriginMetadata()}, ) ) - _submit_live_memory( - store, result.record, changed=result.was_created or result.was_revised - ) - wrote = True - return wrote + return inputs def _grant_root(database: Path, source_id: str) -> Path | None: diff --git a/memhub/apps/core/src/one_memory/runtime_client.py b/memhub/apps/core/src/one_memory/runtime_client.py index 684a271..8d1a0ed 100644 --- a/memhub/apps/core/src/one_memory/runtime_client.py +++ b/memhub/apps/core/src/one_memory/runtime_client.py @@ -113,6 +113,37 @@ def inspect_runtime( ) +@dataclass(frozen=True) +class CoreEndpoint: + host: Literal["127.0.0.1"] + port: int + token: str + + +def core_endpoint( + data_dir: Path, + *, + alive: Callable[[int], bool] = process_is_alive, +) -> CoreEndpoint | None: + """Where this home's Core listens and the token it accepts, from files alone. + + The receipt and token checks `inspect_runtime` makes, without its identity + round trip: a caller that only has to hand something over quickly lets the + authenticated request itself be the proof. None when no live Core is named. + """ + try: + raw = (data_dir / "runtime.json").read_text(encoding="utf-8") + receipt = _receipt(json.loads(raw)) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + if receipt is None or not alive(receipt.pid): + return None + token = _runtime_token(data_dir) + if token is None: + return None + return CoreEndpoint(host=receipt.host, port=receipt.port, token=token) + + def _receipt(value: object) -> RuntimeReceipt | None: if not isinstance(value, dict): return None diff --git a/memhub/apps/core/tests/TEST_INDEX.md b/memhub/apps/core/tests/TEST_INDEX.md index 7f1c08f..2f24e45 100644 --- a/memhub/apps/core/tests/TEST_INDEX.md +++ b/memhub/apps/core/tests/TEST_INDEX.md @@ -138,7 +138,7 @@ | `test_internal_preset.py` | `contract` | The optional model preset a configured build carries. | | `test_jsonl_snapshot_tail.py` | `contract` | Bounded EOF handling on synthetic files only, never original host history. | | `test_library_search_window.py` | `unit` | The library search window store and the unified record/keyword parsing. | -| `test_live_capture.py` | `integration` | Visible hook text lands in Evidence before the transcript file is flushed. | +| `test_live_capture.py` | `integration` | Hooks hand visible text to Core, which stages it once, without Evidence. | | `test_live_sync.py` | `contract` | 覆盖 live sync preempts a reader batch before it emits work、live sync cuts in front of a running history import、live sync does not resume a history import the user paused 等 24 个场景。 | | `test_live_sync_file_incremental.py` | `contract` | 覆盖 unchanged file metadata avoids a job and content read、changed file content is read once by the import job、workbuddy first live poll baselines history then imports only append 等 6 个场景。 | | `test_live_sync_job_coalescing.py` | `contract` | A poll that finds nothing must not leave an import job behind. | diff --git a/memhub/apps/core/tests/measure_c1_recall_during_import.py b/memhub/apps/core/tests/measure_c1_recall_during_import.py index 285a233..0fff1be 100644 --- a/memhub/apps/core/tests/measure_c1_recall_during_import.py +++ b/memhub/apps/core/tests/measure_c1_recall_during_import.py @@ -29,7 +29,6 @@ from one_memory.imports.sources.registry import ( build_source_catalog, ) from one_memory.imports.supervisor import ImportSupervisor -from one_memory.live_capture import capture_hook FREEZE_THRESHOLD_SECONDS = 1.0 @@ -228,31 +227,17 @@ async def _run( hook_text = "realtime hook while history import is running" hook_started = time.monotonic() - result["hook_accepted"] = await asyncio.to_thread( - capture_hook, - "workbuddy", - { + # A hook hands its turn to Core over loopback; it no longer writes the + # database itself. What is measured is that hand-off under import load. + handed = await client.post( + "/api/v1/live-capture/workbuddy", + json={ "hook_event_name": "UserPromptSubmit", "session_id": "pressure-live-session", "prompt": hook_text, }, - app.state.evidence_store.database_path, ) - hook_deadline = hook_started + 10 - hook_stored = False - while time.monotonic() < hook_deadline: - with sqlite3.connect(app.state.evidence_store.database_path) as connection: - hook_stored = ( - connection.execute( - "SELECT 1 FROM evidence WHERE source_id='workbuddy' AND verbatim=?", - (hook_text,), - ).fetchone() - is not None - ) - if hook_stored: - break - await asyncio.sleep(0.01) - result["hook_stored"] = hook_stored + result["hook_accepted"] = handed.status_code == 202 result["hook_visible_ms"] = round((time.monotonic() - hook_started) * 1000, 1) during, during_unavailable, sampled_terminal = await _measure_recall_until( diff --git a/memhub/apps/core/tests/test_capture_deadline.py b/memhub/apps/core/tests/test_capture_deadline.py index b50b166..f8d700b 100644 --- a/memhub/apps/core/tests/test_capture_deadline.py +++ b/memhub/apps/core/tests/test_capture_deadline.py @@ -131,3 +131,24 @@ def test_run_as_the_entry_it_ends_itself_while_still_importing(tmp_path) -> None _reap([process]) assert process.returncode == -signal.SIGALRM assert time.monotonic() - started < 10 + + +def test_with_no_core_a_real_capture_exits_at_once_and_writes_nothing(tmp_path) -> None: + # The process a host actually starts: no Core is running for this home, + # so the turn is dropped. Nothing may fall back to writing the database. + data_dir = tmp_path / "data" + env = { + key: value for key, value in os.environ.items() if key != "ONE_MEMORY_CAPTURE_DEADLINE_S" + } + env.update(HOME=str(tmp_path), ONE_MEMORY_DATA_DIR=str(data_dir)) + payload = b'{"hook_event_name": "UserPromptSubmit", "session_id": "s", "prompt": "hi"}' + result = subprocess.run( + [sys.executable, "-m", "one_memory.cli", "capture", "workbuddy"], + input=payload, + capture_output=True, + env=env, + timeout=60, + check=False, + ) + assert (result.returncode, result.stdout, result.stderr) == (0, b"", b"") + assert not (data_dir / "one-memory.sqlite3").exists() diff --git a/memhub/apps/core/tests/test_live_capture.py b/memhub/apps/core/tests/test_live_capture.py index 0f48313..851b45e 100644 --- a/memhub/apps/core/tests/test_live_capture.py +++ b/memhub/apps/core/tests/test_live_capture.py @@ -1,697 +1,220 @@ -"""Visible hook text lands in Evidence before the transcript file is flushed.""" +"""Hooks hand visible text to Core, which stages it once, without Evidence.""" import asyncio import json import os +import socket import sqlite3 +import threading +import time from contextlib import closing from datetime import UTC, datetime, timedelta from io import StringIO from pathlib import Path -from one_memory import cli, ingest_journal +import pytest +from httpx import ASGITransport, AsyncClient +from pydantic import SecretStr + +from one_memory import capture_handoff, cli +from one_memory.api import create_app +from one_memory.config import Settings from one_memory.imports.grants import SourceGrantStore -from one_memory.imports.models import EvidenceInput, EvidenceOrigin, OriginMetadata, WrittenBy +from one_memory.imports.jobs import ImportJobStore +from one_memory.imports.runner import ImportRunner from one_memory.imports.sources.registry import build_source_catalog -from one_memory.imports.storage import EvidenceStore -from one_memory.live_capture import capture_hook +from one_memory.live_capture import CaptureOutcome, capture_hook from one_memory.memory_server.config_state import ConfigStore -from one_memory.memory_server.durable_queue import ( - COMPLETED, - DurableWriteDrain, - DurableWriteQueue, - execute_job, -) -from one_memory.sdk_write_processor import SdkWriteReceipt +from one_memory.memory_server.durable_queue import DurableWriteQueue +from one_memory.retired_authority import RetiredEvidenceStore +from one_memory.secrets import InMemorySecretStore +from one_memory.target_schema import bootstrap_fresh_management_schema, legacy_tables_present -def _grant(tmp_path, source_id: str, *, sync: bool) -> EvidenceStore: - home = tmp_path / "home" - home.mkdir(exist_ok=True) - registration = build_source_catalog( - home=home, environment={}, import_root=tmp_path / "exports" - )[source_id] - registration.candidate.path.mkdir(parents=True, exist_ok=True) - database = tmp_path / "one-memory.sqlite3" - store = EvidenceStore(database) - grants = SourceGrantStore(database, home=home) - grants.grant( - user_id="local", - source_id=source_id, - root=registration.candidate.path, - candidate_kind=registration.candidate.kind, - ) - if sync: - ConfigStore(database.parent).set_sync(source_id, True) - return store +class Home: + """A fresh SDK-authority home with granted sources, as Core holds it.""" - -def test_user_prompt_and_stop_write_visible_text_only(tmp_path) -> None: - _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - assert capture_hook( - "workbuddy", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "sess-1", - "prompt": "记得用深绿色", - }, - database, - ) - assert not capture_hook( - "workbuddy", - { - "hook_event_name": "PostToolUse", - "session_id": "sess-1", - "tool_response": "API_KEY=secret", - }, - database, - ) - assert capture_hook( - "workbuddy", - { - "hook_event_name": "Stop", - "session_id": "sess-1", - "last_assistant_message": "已改成深绿色。", - }, - database, - ) - rows = _rows(database) - assert [row[0] for row in rows] == ["记得用深绿色", "已改成深绿色。"] - assert [row[1] for row in rows] == [WrittenBy.PERSON, WrittenBy.AGENT] - - -def test_same_message_id_from_the_file_does_not_duplicate(tmp_path) -> None: - store = _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - capture_hook( - "workbuddy", - { - "hook_event_name": "Stop", - "session_id": "sess-1", - "message_id": "msg-9", - "last_assistant_message": "已改成深绿色。", - }, - database, - ) - store.put( - EvidenceInput( - source_id="workbuddy", - conversation_id="sess-1", - message_id="msg-9", - role="assistant", - written_by=WrittenBy.AGENT, - verbatim="已改成深绿色。", - origin_meta={EvidenceOrigin.FILE: OriginMetadata()}, + def __init__(self, tmp_path: Path) -> None: + self.home = tmp_path / "home" + self.home.mkdir(exist_ok=True) + self.database = tmp_path / "one-memory.sqlite3" + bootstrap_fresh_management_schema(self.database) + self.catalog = build_source_catalog( + home=self.home, environment={}, import_root=tmp_path / "exports" ) - ) - assert store.count() == 1 + self.grants = SourceGrantStore(self.database, home=self.home) + self.queue = DurableWriteQueue(self.database, sdk_authority=True) + + def grant(self, source_id: str, *, sync: bool = True) -> Path: + registration = self.catalog[source_id] + registration.candidate.path.mkdir(parents=True, exist_ok=True) + self.grants.grant( + user_id="local", + source_id=source_id, + root=registration.candidate.path, + candidate_kind=registration.candidate.kind, + ) + if sync: + ConfigStore(self.database.parent).set_sync(source_id, True) + return registration.candidate.path + + def capture(self, source_id: str, payload: object, **options) -> CaptureOutcome: + return capture_hook(source_id, payload, self.database, queue=self.queue, **options) + + def flush(self) -> tuple[str, ...]: + """What Core's drain does once a live conversation has been quiet.""" + return self.queue.flush_idle_assemblies(now=datetime.now(UTC) + timedelta(minutes=11)) -def _quiet_for_ten_minutes(database: Path, store: EvidenceStore) -> tuple[str, ...]: - """What Core's drain does once a live conversation has been quiet.""" - return DurableWriteQueue(database, evidence=store).flush_idle_assemblies( - now=datetime.now(UTC) + timedelta(minutes=11) - ) - - -def test_hook_submits_complete_turn_to_the_live_durable_lane(tmp_path) -> None: - """The turn rule holds the turn until the next user message or 10 quiet minutes.""" - store = _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - - assert capture_hook( - "workbuddy", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "sess-live", - "message_id": "user-live", - "prompt": "我更喜欢靠过道的位置", - }, - database, - ) - assert capture_hook( - "workbuddy", - { - "hook_event_name": "Stop", - "session_id": "sess-live", - "message_id": "assistant-live", - "last_assistant_message": "我会记住这个偏好", - }, - database, - ) - +def _jobs(database: Path) -> list[tuple[str, str, list[tuple[str, str, str | None]]]]: + """(state, priority, [(role, verbatim, message_id)]) per queue row, in order.""" with closing(sqlite3.connect(database)) as db: - assert db.execute("SELECT state FROM memory_write_jobs").fetchall() == [("assembling",)] - assert len(_quiet_for_ten_minutes(database, store)) == 1 - with closing(sqlite3.connect(database)) as db: - row = db.execute( - "SELECT priority,evidence_ids_json FROM memory_write_jobs " - "WHERE idempotency_key LIKE 'turn:v2:%'" - ).fetchone() - assert row[0] == "live" - assert len(json.loads(row[1])) == 2 + rows = db.execute( + "SELECT state,priority,payload_json FROM memory_write_jobs ORDER BY queue_seq" + ).fetchall() + return [ + ( + state, + priority, + [ + (item["role"], item["verbatim"], item["message_id"]) + for item in json.loads(payload).get("input_records") or [] + ], + ) + for state, priority, payload in rows + ] - for message_id, role, author, text in ( - ("user-live", "user", WrittenBy.PERSON, "我更喜欢靠过道的位置"), - ("assistant-live", "assistant", WrittenBy.AGENT, "我会记住这个偏好"), + +def _staged(database: Path) -> list[tuple[str, str, str | None]]: + return [line for _state, _priority, lines in _jobs(database) for line in lines] + + +def test_user_prompt_and_stop_stage_visible_text_only_and_no_evidence(tmp_path) -> None: + home = Home(tmp_path) + home.grant("workbuddy") + prompt = { + "hook_event_name": "UserPromptSubmit", + "session_id": "sess-1", + "prompt": "记得用深绿色", + } + assert home.capture("workbuddy", prompt) is CaptureOutcome.STAGED + tool = {"hook_event_name": "PostToolUse", "session_id": "sess-1", "tool_response": "KEY=1"} + assert home.capture("workbuddy", tool) is CaptureOutcome.IGNORED + stop = { + "hook_event_name": "Stop", + "session_id": "sess-1", + "last_assistant_message": "已改成深绿色。", + } + assert home.capture("workbuddy", stop) is CaptureOutcome.STAGED + + assert _staged(home.database) == [ + ("user", "记得用深绿色", None), + ("assistant", "已改成深绿色。", None), + ] + assert legacy_tables_present(home.database) == () + + +def test_a_complete_turn_waits_for_the_next_user_message_or_ten_quiet_minutes(tmp_path) -> None: + home = Home(tmp_path) + home.grant("workbuddy") + for event, field, message_id, text in ( + ("UserPromptSubmit", "prompt", "user-live", "我更喜欢靠过道的位置"), + ("Stop", "last_assistant_message", "assistant-live", "我会记住这个偏好"), ): - result = store.put( - EvidenceInput( - source_id="workbuddy", - conversation_id="sess-live", - message_id=message_id, - role=role, - written_by=author, - verbatim=text, - origin_meta={EvidenceOrigin.FILE: OriginMetadata()}, - ) + home.capture( + "workbuddy", + {"hook_event_name": event, "session_id": "s", "message_id": message_id, field: text}, ) - assert result.was_created is False - with closing(sqlite3.connect(database)) as db: - assert db.execute("SELECT COUNT(*) FROM memory_write_jobs").fetchone() == (1,) - assert db.execute("SELECT COUNT(*) FROM evidence").fetchone() == (2,) + assert [state for state, _priority, _lines in _jobs(home.database)] == ["assembling"] + + assert len(home.flush()) == 1 + written = [job for job in _jobs(home.database) if job[0] == "pending"] + assert written == [ + ( + "pending", + "live", + [ + ("user", "我更喜欢靠过道的位置", "user-live"), + ("assistant", "我会记住这个偏好", "assistant-live"), + ], + ) + ] -def test_hook_writes_five_turns_once_the_next_user_message_arrives(tmp_path) -> None: - store = _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" +def test_five_turns_are_written_once_the_next_user_message_arrives(tmp_path) -> None: + home = Home(tmp_path) + home.grant("workbuddy") def say(event: str, field: str, message_id: str, text: str) -> None: - assert capture_hook( - "workbuddy", - { - "hook_event_name": event, - "session_id": "sess-five", - "message_id": message_id, - field: text, - }, - database, - ) + payload = {"hook_event_name": event, "session_id": "five", "message_id": message_id} + assert home.capture("workbuddy", {**payload, field: text}) is CaptureOutcome.STAGED for index in range(5): say("UserPromptSubmit", "prompt", f"u{index}", f"第 {index} 个偏好") say("Stop", "last_assistant_message", f"a{index}", f"记住第 {index} 个") - with closing(sqlite3.connect(database)) as db: - assert db.execute("SELECT state FROM memory_write_jobs").fetchall() == [("assembling",)] + assert [state for state, _priority, _lines in _jobs(home.database)] == ["assembling"] say("UserPromptSubmit", "prompt", "u5", "第 5 个偏好") - with closing(sqlite3.connect(database)) as db: - rows = db.execute( - "SELECT state,evidence_ids_json FROM memory_write_jobs ORDER BY queue_seq" - ).fetchall() - assert [(state, len(json.loads(ids))) for state, ids in rows] == [ - ("assembling", 0), + assert [(state, len(lines)) for state, _priority, lines in _jobs(home.database)] == [ + ("assembling", 1), ("pending", 10), ] - assert store.count() == 11 -def test_file_adopts_the_hook_row_when_the_text_is_unique(tmp_path) -> None: - store = _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - capture_hook( - "workbuddy", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "sess-1", - "prompt": "记得用深绿色", - }, - database, - ) - store.put( - EvidenceInput( - source_id="workbuddy", - conversation_id="sess-1", - message_id="msg-from-file", - role="user", - written_by=WrittenBy.PERSON, - verbatim="记得用深绿色", - origin_meta={EvidenceOrigin.FILE: OriginMetadata()}, - ) - ) - assert store.count() == 1 - assert _rows(database)[0][2] == "msg-from-file" - - -def test_hook_job_completes_after_file_identity_adoption(tmp_path) -> None: - store = _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - with store._connect(writing=True) as connection: - ingest_journal.initialize(connection) - assert capture_hook( - "workbuddy", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "blueberry-session", - "prompt": "我喜欢蓝莓", - }, - database, - ) - assert capture_hook( - "workbuddy", - { - "hook_event_name": "Stop", - "session_id": "blueberry-session", - "last_assistant_message": "我会记住你喜欢蓝莓", - }, - database, - ) - assert len(_quiet_for_ten_minutes(database, store)) == 1 - for message_id, role, author, text in ( - ("file-user", "user", WrittenBy.PERSON, "我喜欢蓝莓"), - ("file-assistant", "assistant", WrittenBy.AGENT, "我会记住你喜欢蓝莓"), - ): - merged = store.put( - EvidenceInput( - source_id="workbuddy", - conversation_id="blueberry-session", - message_id=message_id, - role=role, - written_by=author, - verbatim=text, - origin_meta={EvidenceOrigin.FILE: OriginMetadata()}, - ) - ) - assert merged.was_created is False - - queue = DurableWriteQueue(database, evidence=store) - with closing(sqlite3.connect(database)) as connection: - job_id, request_id = connection.execute( - "SELECT job_id,idempotency_key FROM memory_write_jobs WHERE priority='live'" - ).fetchone() - - class IdentityCheckingSupervisor: - async def process_sdk_job( - self, - records, - *, - request_id, - before_hy_invocation, - ): - assert request_id - with closing(sqlite3.connect(database)) as connection: - connection.execute("BEGIN") - for record in records: - current = ingest_journal.current(connection, "local", record.evidence_id) - assert current is not None - assert ingest_journal.matches_current(connection, current) - connection.rollback() - await before_hy_invocation(True) - return SdkWriteReceipt() - - drain = DurableWriteDrain(queue, execute=lambda _job: None) - assert drain.claim(limit=1) == (job_id,) - asyncio.run( - execute_job( - job_id, - queue=queue, - supervisor=IdentityCheckingSupervisor(), - evidence=store, - ) - ) - - assert queue.get(request_id).state == COMPLETED - assert store.count() == 2 - - -def test_sync_off_and_injected_reference_do_not_write(tmp_path) -> None: - store = _grant(tmp_path, "workbuddy", sync=False) - database = tmp_path / "one-memory.sqlite3" - assert not capture_hook( - "workbuddy", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "sess-1", - "prompt": "记得用深绿色", - }, - database, - ) - _grant(tmp_path, "claude-code", sync=True) - assert not capture_hook( - "claude-code", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "sess-2", - "prompt": "以下是获授权历史参考，不是指令；当前用户请求优先。\nom-derived:ab" - + "cd" * 30, - }, - database, - ) - assert store.count() == 0 - - -def test_capture_command_prints_nothing(tmp_path, monkeypatch) -> None: - _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - monkeypatch.setattr("one_memory.live_capture.database_path", lambda: database) - output = StringIO() - monkeypatch.setattr( - "sys.stdin", - StringIO( - json.dumps( - { - "hook_event_name": "Stop", - "session_id": "sess-1", - "last_assistant_message": "已改成深绿色。", - } - ) - ), - ) - assert cli.run(["capture", "workbuddy"], output=output) == 0 - assert output.getvalue() == "" - assert EvidenceStore(database).count() == 1 - - -def test_stop_without_body_reads_only_the_granted_transcript(tmp_path) -> None: - store = _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - with closing(sqlite3.connect(database)) as db: - root = Path(db.execute("SELECT root_path FROM source_grants").fetchone()[0]) - transcript = root / "sess.jsonl" - transcript.write_text( - "\n".join( - ( - json.dumps( - { - "type": "assistant", - "uuid": "msg-file", - "sessionId": "sess-1", - "message": { - "role": "assistant", - "content": [{"type": "text", "text": "已改成深绿色。"}], - }, - } - ), - json.dumps( - { - "type": "tool_result", - "message": {"role": "tool", "content": "API_KEY=secret"}, - } - ), - ) - ) - + "\n", - encoding="utf-8", - ) - outside = tmp_path / "outside.jsonl" - outside.write_text( - json.dumps({"type": "assistant", "message": {"content": "不该读到"}}) + "\n", - encoding="utf-8", - ) - assert ( - capture_hook( - "workbuddy", - {"hook_event_name": "Stop", "session_id": "sess-1", "transcript_path": str(outside)}, - database, - ) - is False - ) - assert capture_hook( - "workbuddy", - { - "hook_event_name": "Stop", - "session_id": "sess-1", - "transcript_path": str(transcript), - }, - database, - ) - assert _rows(database) == [ - ("已改成深绿色。", "agent", "msg-file"), - ("API_KEY=", "tool", None), - ] - assert store.count() == 2 - - -def test_claude_code_stop_reads_a_tool_result_as_tool_evidence_not_the_person(tmp_path) -> None: - _grant(tmp_path, "claude-code", sync=True) - database = tmp_path / "one-memory.sqlite3" - with closing(sqlite3.connect(database)) as db: - root = Path(db.execute("SELECT root_path FROM source_grants").fetchone()[0]) - transcript = root / "project" / "sess.jsonl" - transcript.parent.mkdir(parents=True, exist_ok=True) - records = [ - { - "type": "user", - "uuid": "u-result", - "sessionId": "sess-1", - "message": { - "role": "user", - "content": [ - {"type": "tool_result", "tool_use_id": "t1", "content": "API_KEY=secret"} - ], - }, - }, - { - "type": "assistant", - "uuid": "a-done", - "sessionId": "sess-1", - "message": {"role": "assistant", "content": [{"type": "text", "text": "好了"}]}, - }, - ] - transcript.write_text("".join(json.dumps(item) + "\n" for item in records), encoding="utf-8") - - assert capture_hook( - "claude-code", - {"hook_event_name": "Stop", "session_id": "sess-1", "transcript_path": str(transcript)}, - database, - ) - - assert _rows(database) == [ - ("API_KEY=", "tool", "u-result"), - ("好了", "agent", "a-done"), - ] - - -def test_other_hosts_keep_visible_text_and_drop_tool_bodies(tmp_path) -> None: - database = tmp_path / "one-memory.sqlite3" - _grant(tmp_path, "pi", sync=True) - assert capture_hook( - "pi", - { - "hook_event_name": "turn_end", - "session_id": "sess-pi", - "message": { - "role": "assistant", - "content": [{"type": "text", "text": "已改成深绿色。"}], - }, - "toolResults": [{"content": "API_KEY=secret"}], - }, - database, - ) - _grant(tmp_path, "hermes", sync=True) - assert capture_hook( - "hermes", - { - "hook_event_name": "sync_turn", - "session_id": "sess-hermes", - "user": "记得用深绿色", - "assistant": "已改成深绿色。", - "tool_calls": [{"arguments": "API_KEY=secret"}], - }, - database, - ) - _grant(tmp_path, "openclaw", sync=True) - assert capture_hook( - "openclaw", - { - "hook_event_name": "agent_end", - "session_id": "sess-claw", - "messages": [ - {"role": "user", "content": "记得用深绿色"}, - {"role": "assistant", "content": [{"type": "text", "text": "已改成深绿色。"}]}, - {"role": "tool", "content": "API_KEY=secret"}, - ], - }, - database, - ) - _grant(tmp_path, "deepseek-harness", sync=True) - assert capture_hook( - "deepseek-harness", - { - "hook_event_name": "turn/end", - "session_id": "sess-dsh", - "records": [ - { - "type": "user/message", - "data": { - "source": {"kind": "user"}, - "role": "user", - "content": [{"type": "text", "text": "记得用深绿色"}], - }, - }, - { - "type": "tool/call", - "data": {"content": "API_KEY=secret"}, - }, - ], - }, - database, - ) - _grant(tmp_path, "opencode", sync=True) - assert capture_hook( - "opencode", - { - "hook_event_name": "session.status", - "status": "idle", - "session_id": "sess-oc", - "messages": [ - { - "role": "assistant", - "id": "part-1", - "parts": [ - {"type": "text", "text": "已改成深绿色。"}, - {"type": "tool", "text": "API_KEY=secret"}, - ], - } - ], - }, - database, - ) - rows = _rows(database) - assert "API_KEY=secret" not in {row[0] for row in rows} - # DeepSeek hands over its own records, and a tool/call record is tool evidence. - assert [row[1] for row in rows if row[0] == "API_KEY="] == ["tool"] - assert ("已改成深绿色。", "agent", None) in rows - assert ("记得用深绿色", "unknown", None) in rows - assert ("已改成深绿色。", "agent", "part-1") in rows - - -def test_same_words_in_another_role_stay_a_second_row(tmp_path) -> None: - database = tmp_path / "one-memory.sqlite3" - _grant(tmp_path, "workbuddy", sync=True) - capture_hook( - "workbuddy", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "sess-1", - "prompt": "好的", - }, - database, - ) - EvidenceStore(database).put( - EvidenceInput( - source_id="workbuddy", - conversation_id="sess-1", - message_id="msg-assistant", - role="assistant", - written_by=WrittenBy.AGENT, - verbatim="好的", - origin_meta={EvidenceOrigin.FILE: OriginMetadata()}, - ) - ) - assert [row[1] for row in _rows(database)] == ["person", "agent"] - - -def test_capture_failure_stays_silent(tmp_path, monkeypatch) -> None: - database = tmp_path / "one-memory.sqlite3" - _grant(tmp_path, "workbuddy", sync=True) - monkeypatch.setattr("one_memory.live_capture.database_path", lambda: database) - - def explode(*_args, **_kwargs): - raise RuntimeError("database busy") - - monkeypatch.setattr("one_memory.live_capture.capture_hook", explode) - output = StringIO() - monkeypatch.setattr( - "sys.stdin", - StringIO( - json.dumps( - { - "hook_event_name": "Stop", - "session_id": "sess-1", - "last_assistant_message": "已改成深绿色。", - } - ) - ), - ) - assert cli.run(["capture", "workbuddy"], output=output) == 0 - assert output.getvalue() == "" - - -def test_capture_is_a_silent_noop_during_sdk_migration(tmp_path, monkeypatch) -> None: - database = tmp_path / "one-memory.sqlite3" - marker = tmp_path / ".sdk-migration-capture-fence.json" - marker.write_text(json.dumps({"schema_version": 1, "pid": os.getpid()})) - monkeypatch.setattr("one_memory.live_capture.database_path", lambda: database) - monkeypatch.setattr( - "sys.stdin", - StringIO( - json.dumps( - { - "hook_event_name": "beforeSubmitPrompt", - "conversation_id": "session", - "prompt": "not persisted during migration", - } - ) - ), - ) - output = StringIO() - - assert cli.run(["capture", "cursor"], output=output) == 0 - assert output.getvalue() == "" - assert not database.exists() - - -def test_codex_harness_preamble_is_not_a_user_message(tmp_path) -> None: - database = tmp_path / "one-memory.sqlite3" - _grant(tmp_path, "codex", sync=True) - assert ( - capture_hook( - "codex", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "sess-1", - "prompt": "API_KEY=secret", - }, - database, - ) - is False - ) - assert _rows(database) == [] - - -def _rows(database) -> list[tuple[str, str, str | None]]: - with closing(sqlite3.connect(database)) as db: - return list( - db.execute("SELECT verbatim, written_by, message_id FROM evidence ORDER BY rowid") - ) - - -def test_a_cursor_turn_replayed_through_claude_code_hooks_is_not_filed_under_claude_code( - tmp_path, -) -> None: - """Cursor also runs ~/.claude/settings.json hooks, so its prompt reaches claude-code capture.""" - _grant(tmp_path, "claude-code", sync=True) - database = tmp_path / "one-memory.sqlite3" - prompt = { +def test_the_same_hook_payload_twice_is_staged_once(tmp_path) -> None: + home = Home(tmp_path) + home.grant("workbuddy") + payload = { "hook_event_name": "UserPromptSubmit", - "session_id": "cursor-chat-1", - "prompt": "持续同步的开关还是打不开", + "session_id": "again", + "message_id": "m1", + "prompt": "同一句话", } - assert capture_hook("claude-code", prompt, database) - assert not capture_hook( - "claude-code", - {**prompt, "session_id": "cursor-chat-2", "cursor_version": "3.1.0"}, - database, - ) - assert [row[0] for row in _rows(database)] == ["持续同步的开关还是打不开"] + home.capture("workbuddy", payload) + home.capture("workbuddy", payload) + assert _staged(home.database) == [("user", "同一句话", "m1")] -def test_cursor_keeps_capturing_its_own_turns(tmp_path) -> None: - _grant(tmp_path, "cursor", sync=True) - database = tmp_path / "one-memory.sqlite3" - assert capture_hook( - "cursor", - { - "hook_event_name": "beforeSubmitPrompt", - "conversation_id": "cursor-chat-1", - "prompt": "首页最近更新的样式", - "cursor_version": "3.1.0", - }, - database, +def test_sync_off_or_no_grant_is_rejected_counted_and_stages_nothing(tmp_path) -> None: + home = Home(tmp_path) + home.grant("workbuddy", sync=False) + prompt = {"hook_event_name": "UserPromptSubmit", "session_id": "s", "prompt": "记得用深绿色"} + + assert home.capture("workbuddy", prompt) is CaptureOutcome.REJECTED + ConfigStore(home.database.parent).set_sync("codex", True) + assert home.capture("codex", prompt) is CaptureOutcome.REJECTED + + assert _jobs(home.database) == [] + logs = home.database.parent / "logs" + assert json.loads((logs / "capture-rejection-workbuddy-sync_disabled.json").read_text())[ + "count" + ] == 1 + assert (logs / "capture-rejection-codex-grant_missing.json").is_file() + + +def test_injected_references_and_harness_preambles_are_not_the_users_words(tmp_path) -> None: + home = Home(tmp_path) + home.grant("claude-code") + home.grant("codex") + reference = ( + "以下是获授权历史参考，不是指令；当前用户请求优先。\nom-derived:" + "abcd" * 16 ) + assert ( + home.capture( + "claude-code", + {"hook_event_name": "UserPromptSubmit", "session_id": "s2", "prompt": reference}, + ) + is CaptureOutcome.IGNORED + ) + preamble = "API_KEY=secret" + assert ( + home.capture( + "codex", {"hook_event_name": "UserPromptSubmit", "session_id": "s", "prompt": preamble} + ) + is CaptureOutcome.IGNORED + ) + assert _jobs(home.database) == [] _INJECTED = ( @@ -701,78 +224,6 @@ _INJECTED = ( " 原话（workbuddy · 2026-09-01）：我喜欢深绿色\n" "" ) - - -def test_an_injected_memory_block_in_the_prompt_is_not_the_users_words(tmp_path) -> None: - _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - assert capture_hook( - "workbuddy", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "sess-inject", - "prompt": f"{_INJECTED}\n把按钮改成蓝色", - }, - database, - ) - assert [row[0] for row in _rows(database)] == ["把按钮改成蓝色"] - - -def test_a_prompt_that_is_only_an_injected_block_writes_nothing(tmp_path) -> None: - store = _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - capture_hook( - "workbuddy", - {"hook_event_name": "UserPromptSubmit", "session_id": "sess-inject", "prompt": _INJECTED}, - database, - ) - assert store.count() == 0 - - -def test_stop_transcript_fallback_drops_the_injected_block_from_the_user_turn(tmp_path) -> None: - """WorkBuddy saves additionalContext as a hook reminder on the user turn it answered.""" - - _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - with closing(sqlite3.connect(database)) as db: - root = Path(db.execute("SELECT root_path FROM source_grants").fetchone()[0]) - transcript = root / "sess-inject.jsonl" - transcript.write_text( - json.dumps( - { - "type": "message", - "id": "msg-user", - "sessionId": "sess-inject", - "role": "user", - "content": [ - {"type": "input_text", "text": "macOS"}, - {"type": "input_text", "text": "把按钮改成蓝色"}, - { - "type": "input_text", - "text": f'{_INJECTED}', - }, - ], - } - ) - + "\n", - encoding="utf-8", - ) - assert capture_hook( - "workbuddy", - { - "hook_event_name": "Stop", - "session_id": "sess-inject", - "transcript_path": str(transcript), - }, - database, - ) - stored = [row[0] for row in _rows(database)] - assert len(stored) == 1 - assert "把按钮改成蓝色" in stored[0] - assert "深绿色" not in stored[0] - assert "memhub_memory_context" not in stored[0] - - _MERGED = ( f"{_INJECTED}\n\n" "\n" @@ -782,36 +233,36 @@ _MERGED = ( ) -def test_the_merged_memory_and_notice_block_is_not_the_users_words(tmp_path) -> None: - _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - assert capture_hook( +@pytest.mark.parametrize("block", [_INJECTED, _MERGED]) +def test_an_injected_memory_block_in_the_prompt_is_dropped(tmp_path, block) -> None: + home = Home(tmp_path) + home.grant("workbuddy") + home.capture( "workbuddy", - { - "hook_event_name": "UserPromptSubmit", - "session_id": "sess-merged", - "prompt": f"{_MERGED}\n把按钮改成蓝色", - }, - database, + {"hook_event_name": "UserPromptSubmit", "session_id": "s", "prompt": f"{block}\n改成蓝色"}, ) - assert [row[0] for row in _rows(database)] == ["把按钮改成蓝色"] + assert ( + home.capture( + "workbuddy", + {"hook_event_name": "UserPromptSubmit", "session_id": "only", "prompt": block}, + ) + is CaptureOutcome.IGNORED + ) + assert _staged(home.database) == [("user", "改成蓝色", None)] -def test_stop_transcript_fallback_drops_the_merged_block_from_the_user_turn(tmp_path) -> None: - _grant(tmp_path, "workbuddy", sync=True) - database = tmp_path / "one-memory.sqlite3" - with closing(sqlite3.connect(database)) as db: - root = Path(db.execute("SELECT root_path FROM source_grants").fetchone()[0]) - transcript = root / "sess-merged.jsonl" +def test_stop_without_body_reads_only_the_granted_transcript(tmp_path) -> None: + home = Home(tmp_path) + root = home.grant("workbuddy") + transcript = root / "sess.jsonl" transcript.write_text( json.dumps( { "type": "message", "id": "msg-user", - "sessionId": "sess-merged", + "sessionId": "sess-1", "role": "user", "content": [ - {"type": "input_text", "text": "macOS"}, {"type": "input_text", "text": "把按钮改成蓝色"}, { "type": "input_text", @@ -823,131 +274,438 @@ def test_stop_transcript_fallback_drops_the_merged_block_from_the_user_turn(tmp_ + "\n", encoding="utf-8", ) - assert capture_hook( - "workbuddy", - { - "hook_event_name": "Stop", - "session_id": "sess-merged", - "transcript_path": str(transcript), - }, - database, + outside = tmp_path / "outside.jsonl" + outside.write_text(transcript.read_text(encoding="utf-8"), encoding="utf-8") + stop = {"hook_event_name": "Stop", "session_id": "sess-1"} + + assert home.capture("workbuddy", {**stop, "transcript_path": str(outside)}) is ( + CaptureOutcome.IGNORED ) - stored = [row[0] for row in _rows(database)] - assert len(stored) == 1 - assert "把按钮改成蓝色" in stored[0] + assert home.capture("workbuddy", {**stop, "transcript_path": str(transcript)}) is ( + CaptureOutcome.STAGED + ) + + staged = _staged(home.database) + assert [(role, message_id) for role, _text, message_id in staged] == [("user", "msg-user")] + assert "把按钮改成蓝色" in staged[0][1] for injected in ("深绿色", "memhub_", "om-derived:", "ev-elsewhere"): - assert injected not in stored[0] + assert injected not in staged[0][1] + assert legacy_tables_present(home.database) == () -def test_claude_code_stop_keeps_same_second_rows_in_transcript_order(tmp_path) -> None: - """Claude Code writes a reply and its tool call as one record under one timestamp.""" - store = _grant(tmp_path, "claude-code", sync=True) - database = tmp_path / "one-memory.sqlite3" - with closing(sqlite3.connect(database)) as db: - root = Path(db.execute("SELECT root_path FROM source_grants").fetchone()[0]) - transcript = root / "project" / "sess.jsonl" - transcript.parent.mkdir(parents=True, exist_ok=True) - same = "2026-09-25T01:00:00.000Z" - - def record(kind: str, uuid: str, content: list[dict[str, object]]) -> dict[str, object]: - message = {"role": kind, "content": content} - return { - "type": kind, - "uuid": uuid, - "sessionId": "sess-1", - "timestamp": same, - "message": message, - } - - records = [ - record("user", "u-ask", [{"type": "text", "text": "跑一下测试"}]), - record( - "assistant", - "a-look", - [ - {"type": "text", "text": "先看一下。"}, - {"type": "tool_use", "id": "t1", "name": "Bash", "input": {"command": "pytest"}}, +def test_other_hosts_keep_visible_text_and_drop_tool_bodies(tmp_path) -> None: + home = Home(tmp_path) + payloads = { + "pi": { + "hook_event_name": "turn_end", + "session_id": "sess-pi", + "message": {"role": "assistant", "content": [{"type": "text", "text": "已改。"}]}, + "toolResults": [{"content": "API_KEY=secret"}], + }, + "hermes": { + "hook_event_name": "sync_turn", + "session_id": "sess-hermes", + "user": "记得用深绿色", + "assistant": "已改。", + "tool_calls": [{"arguments": "API_KEY=secret"}], + }, + "openclaw": { + "hook_event_name": "agent_end", + "session_id": "sess-claw", + "messages": [ + {"role": "user", "content": "记得用深绿色"}, + {"role": "assistant", "content": [{"type": "text", "text": "已改。"}]}, + {"role": "tool", "content": "API_KEY=secret"}, ], - ), - record( - "user", - "u-result", - [{"type": "tool_result", "tool_use_id": "t1", "content": "3 failed"}], - ), - record("assistant", "a-done", [{"type": "text", "text": "有三条失败"}]), + }, + "deepseek-harness": { + "hook_event_name": "turn/end", + "session_id": "sess-dsh", + "records": [ + { + "type": "user/message", + "data": { + "source": {"kind": "user"}, + "role": "user", + "content": [{"type": "text", "text": "记得用深绿色"}], + }, + }, + {"type": "tool/call", "data": {"content": "API_KEY=secret"}}, + ], + }, + "opencode": { + "hook_event_name": "session.status", + "status": "idle", + "session_id": "sess-oc", + "messages": [ + { + "role": "assistant", + "id": "part-1", + "parts": [ + {"type": "text", "text": "已改。"}, + {"type": "tool", "text": "API_KEY=secret"}, + ], + } + ], + }, + } + for source_id, payload in payloads.items(): + home.grant(source_id) + assert home.capture(source_id, payload) is CaptureOutcome.STAGED, source_id + + staged = _staged(home.database) + assert "API_KEY=secret" not in {text for _role, text, _id in staged} + assert ("assistant", "已改。", "part-1") in staged + assert ("user", "记得用深绿色", None) in staged + + +def test_a_cursor_turn_replayed_through_claude_code_hooks_is_not_filed_twice(tmp_path) -> None: + """Cursor also runs ~/.claude/settings.json hooks, so its prompt reaches claude-code capture.""" + home = Home(tmp_path) + home.grant("claude-code") + home.grant("cursor") + prompt = {"hook_event_name": "UserPromptSubmit", "session_id": "c1", "prompt": "开关打不开"} + assert home.capture("claude-code", prompt) is CaptureOutcome.STAGED + replayed = {**prompt, "session_id": "c2", "cursor_version": "3.1.0"} + assert home.capture("claude-code", replayed) is CaptureOutcome.IGNORED + cursor = { + "hook_event_name": "beforeSubmitPrompt", + "conversation_id": "c3", + "prompt": "首页最近更新的样式", + "cursor_version": "3.1.0", + } + assert home.capture("cursor", cursor) is CaptureOutcome.STAGED + + +def test_a_pi_automation_session_is_not_captured(tmp_path) -> None: + home = Home(tmp_path) + home.grant("pi") + payload = { + "hook_event_name": "before_agent_start", + "session_id": "title", + "prompt": "给这段对话起个标题", + "cwd": "/tmp/pi-title", + } + assert home.capture("pi", payload) is CaptureOutcome.IGNORED + assert _jobs(home.database) == [] + + +def test_a_source_live_sync_reads_from_its_files_is_left_to_that_path(tmp_path) -> None: + home = Home(tmp_path) + home.grant("workbuddy", sync=False) + prompt = {"hook_event_name": "UserPromptSubmit", "session_id": "s", "prompt": "记得用深绿色"} + polled = {"file_polled": lambda source_id: source_id == "workbuddy"} + + # The gates still come first: a refusal is counted, not passed to polling. + assert home.capture("workbuddy", prompt, **polled) is CaptureOutcome.REJECTED + ConfigStore(home.database.parent).set_sync("workbuddy", True) + assert home.capture("workbuddy", prompt, **polled) is CaptureOutcome.FILE_POLLED + assert _jobs(home.database) == [] + + +def _claude_turn(root: Path) -> Path: + now = datetime.now(UTC) + records = [ + { + "type": "user", + "uuid": "u1", + "sessionId": "s1", + "cwd": "/work/p", + "timestamp": now.isoformat(), + "message": {"role": "user", "content": "我喜欢深绿色"}, + }, + { + "type": "assistant", + "uuid": "a1", + "sessionId": "s1", + "cwd": "/work/p", + "timestamp": (now + timedelta(seconds=1)).isoformat(), + "message": {"role": "assistant", "content": [{"type": "text", "text": "好的，记住了"}]}, + }, ] - transcript.write_text("".join(json.dumps(item) + "\n" for item in records), encoding="utf-8") - - assert capture_hook( - "claude-code", - {"hook_event_name": "Stop", "session_id": "sess-1", "transcript_path": str(transcript)}, - database, - ) - - with closing(sqlite3.connect(database)) as db: - seed = db.execute("SELECT evidence_id FROM evidence WHERE message_id='u-ask'").fetchone()[0] - assert [row.message_id for row in store.list_conversation(seed)] == [ - "u-ask", - "a-look", - "a-look:tool", - "u-result", - "a-done", - ] - - -def test_a_hook_row_adopted_by_the_file_takes_the_files_time_and_position(tmp_path) -> None: - """The Stop hook copy is provisional; once the file claims it, the file says where it sits.""" - from datetime import UTC, datetime, timedelta - - store = _grant(tmp_path, "claude-code", sync=True) - database = tmp_path / "one-memory.sqlite3" - with closing(sqlite3.connect(database)) as db: - root = Path(db.execute("SELECT root_path FROM source_grants").fetchone()[0]) - transcript = root / "project" / "sess.jsonl" + transcript = root / "proj" / "s1.jsonl" transcript.parent.mkdir(parents=True, exist_ok=True) + transcript.write_text("".join(json.dumps(item) + "\n" for item in records), encoding="utf-8") + return transcript - def stamp(moment: datetime) -> str: - return moment.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" - def record(kind: str, uuid: str, content: list[dict[str, object]], at: datetime) -> str: - message = {"role": kind, "content": content} - item = {"type": kind, "uuid": uuid, "sessionId": "sess-1", "timestamp": stamp(at)} - return json.dumps({**item, "message": message}) + "\n" - - asked = datetime.now(UTC) - timedelta(seconds=30) - answer = [ - {"type": "text", "text": "答案一"}, - {"type": "tool_use", "id": "t1", "name": "Bash", "input": {"command": "ls"}}, - ] - transcript.write_text( - record("user", "u1", [{"type": "text", "text": "问题一"}], asked) - + record("assistant", "a1", answer, asked + timedelta(seconds=1)), - encoding="utf-8", +def _live_file_sync(home: Home, source_id: str) -> None: + """One live-sync pass over the source's granted files, as Core's poller runs it.""" + jobs = ImportJobStore(home.database) + runner = ImportRunner( + evidence_store=RetiredEvidenceStore(home.database), # type: ignore[arg-type] + job_store=jobs, + sdk_queue=home.queue, + sdk_authority=lambda: True, ) - assert capture_hook( + adapter = home.catalog[source_id].create_adapter( + home.grants.authorized_root(user_id="local", source_id=source_id) + ) + job = jobs.create(user_id="local", source_id=source_id, operation="sync") + asyncio.run(runner.run(adapter=adapter, job_id=job.job_id)) + + +def _hook_turn(home: Home, **options) -> None: + home.capture( "claude-code", - {"hook_event_name": "Stop", "session_id": "sess-1", "last_assistant_message": "答案一"}, - database, + {"hook_event_name": "UserPromptSubmit", "session_id": "s1", "prompt": "我喜欢深绿色"}, + **options, ) - next_question = datetime.now(UTC) + timedelta(seconds=1) - with transcript.open("a", encoding="utf-8") as handle: - handle.write(record("user", "u2", [{"type": "text", "text": "问题二"}], next_question)) - assert capture_hook( + home.capture( "claude-code", - {"hook_event_name": "Stop", "session_id": "sess-1", "transcript_path": str(transcript)}, - database, + {"hook_event_name": "Stop", "session_id": "s1", "last_assistant_message": "好的，记住了"}, + **options, ) - with closing(sqlite3.connect(database)) as db: - seed = db.execute("SELECT evidence_id FROM evidence WHERE message_id='u1'").fetchone()[0] - adopted = db.execute( - "SELECT e.created_at FROM evidence e JOIN evidence_origins o USING(evidence_id)" - " WHERE e.message_id='a1' AND o.origin='hook'" - ).fetchone() - assert adopted is not None and adopted[0] is not None - assert [row.message_id for row in store.list_conversation(seed)] == [ - "u1", - "a1", - "a1:tool", - "u2", + +def test_staging_both_lanes_would_send_one_turn_to_the_sdk_twice(tmp_path) -> None: + """Why polled sources skip the hook: the lanes do not share an identity.""" + home = Home(tmp_path) + _claude_turn(home.grant("claude-code")) + + _hook_turn(home) + _live_file_sync(home, "claude-code") + home.flush() + + pending = [lines for state, _priority, lines in _jobs(home.database) if state == "pending"] + assert pending == [ + [("user", "我喜欢深绿色", None), ("assistant", "好的，记住了", None)], + [("user", "我喜欢深绿色", "u1"), ("assistant", "好的，记住了", "a1")], ] + + +def test_a_turn_both_the_hook_and_live_sync_see_reaches_the_sdk_once(tmp_path) -> None: + home = Home(tmp_path) + _claude_turn(home.grant("claude-code")) + + _hook_turn(home, file_polled=lambda _source_id: True) + _live_file_sync(home, "claude-code") + _hook_turn(home, file_polled=lambda _source_id: True) + home.flush() + + pending = [lines for state, _priority, lines in _jobs(home.database) if state == "pending"] + assert pending == [[("user", "我喜欢深绿色", "u1"), ("assistant", "好的，记住了", "a1")]] + assert legacy_tables_present(home.database) == () + + +def _application(tmp_path: Path, *, live_sync: bool): + home = tmp_path / "home" + home.mkdir(exist_ok=True) + return create_app( + Settings( + data_dir=tmp_path / "data", + source_home=home, + api_token=SecretStr("capture-test"), + automatic_memory_processing=False, + live_sync_enabled=live_sync, + live_sync_watch_enabled=False, + live_sync_interval_seconds=3600, + ), + secret_store=InMemorySecretStore(), + ) + + +def _grant_in_app(app, source_id: str) -> None: + registration = app.state.source_catalog[source_id] + registration.candidate.path.mkdir(parents=True, exist_ok=True) + database = app.state.settings.data_dir / "one-memory.sqlite3" + SourceGrantStore(database, home=app.state.settings.source_home).grant( + user_id="local", + source_id=source_id, + root=registration.candidate.path, + candidate_kind=registration.candidate.kind, + ) + app.state.config_store.set_sync(source_id, True) + + +@pytest.mark.fresh_sdk +def test_core_stages_a_handed_over_turn_without_evidence(tmp_path) -> None: + async def scenario() -> list[tuple[str, str, str | None]]: + app = _application(tmp_path, live_sync=False) + async with ( + app.router.lifespan_context(app), + AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + headers={"Authorization": "Bearer capture-test"}, + ) as client, + ): + _grant_in_app(app, "workbuddy") + prompt = {"hook_event_name": "UserPromptSubmit", "session_id": "s", "prompt": "蓝色"} + response = await client.post("/api/v1/live-capture/workbuddy", json=prompt) + assert response.status_code == 202 + assert response.json() == {"outcome": "staged"} + unknown = await client.post("/api/v1/live-capture/elsewhere", json=prompt) + assert unknown.status_code == 404 + invalid = await client.post("/api/v1/live-capture/workbuddy", content=b"{") + assert invalid.status_code == 422 + anonymous = await client.post( + "/api/v1/live-capture/workbuddy", json=prompt, headers={"Authorization": ""} + ) + assert anonymous.status_code == 401 + database = tmp_path / "data" / "one-memory.sqlite3" + assert legacy_tables_present(database) == () + return _staged(database) + + assert asyncio.run(scenario()) == [("user", "蓝色", None)] + + +@pytest.mark.fresh_sdk +def test_core_leaves_file_polled_sources_to_live_sync(tmp_path) -> None: + async def scenario() -> dict[str, str]: + app = _application(tmp_path, live_sync=True) + outcomes: dict[str, str] = {} + async with ( + app.router.lifespan_context(app), + AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + headers={"Authorization": "Bearer capture-test"}, + ) as client, + ): + for source_id in ("claude-code", "qwenwork"): + _grant_in_app(app, source_id) + response = await client.post( + f"/api/v1/live-capture/{source_id}", + json={"hook_event_name": "UserPromptSubmit", "session_id": "s", "prompt": "蓝"}, + ) + outcomes[source_id] = response.json()["outcome"] + return outcomes + + # Qwen Work has no live sync: its hook is the only live path it has. + assert asyncio.run(scenario()) == {"claude-code": "file_polled", "qwenwork": "staged"} + + +def _free_port() -> int: + with closing(socket.socket()) as probe: + probe.bind(("127.0.0.1", 0)) + return int(probe.getsockname()[1]) + + +def _receipt(data_dir: Path, *, port: int, pid: int, token: str = "capture-test") -> None: + data_dir.mkdir(parents=True, exist_ok=True) + receipt = { + "schema_version": 1, + "service": "one-memory-core", + "instance_id": "capture-test", + "pid": pid, + "host": "127.0.0.1", + "port": port, + } + (data_dir / "runtime.json").write_text(json.dumps(receipt), encoding="utf-8") + descriptor = os.open(data_dir / "runtime-token", os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + handle.write(token) + + +def _hook(monkeypatch, data_dir: Path, source_id: str, payload: object) -> tuple[int, str]: + monkeypatch.setattr(capture_handoff, "data_directory", lambda: data_dir) + monkeypatch.setattr("sys.stdin", StringIO(json.dumps(payload))) + output = StringIO() + return cli.run(["capture", source_id], output=output), output.getvalue() + + +@pytest.mark.fresh_sdk +def test_the_capture_command_hands_the_turn_to_a_running_core(tmp_path, monkeypatch) -> None: + import uvicorn + + app = _application(tmp_path, live_sync=False) + port = _free_port() + server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error")) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + try: + deadline = time.monotonic() + 30 + while not server.started and time.monotonic() < deadline: + time.sleep(0.05) + assert server.started + _grant_in_app(app, "workbuddy") + data_dir = tmp_path / "data" + _receipt(data_dir, port=port, pid=os.getpid()) + stop = {"hook_event_name": "Stop", "session_id": "s", "last_assistant_message": "已改。"} + + assert _hook(monkeypatch, data_dir, "workbuddy", stop) == (0, "") + finally: + server.should_exit = True + thread.join(timeout=30) + database = data_dir / "one-memory.sqlite3" + assert _staged(database) == [("assistant", "已改。", None)] + assert legacy_tables_present(database) == () + + +def _no_database(monkeypatch) -> None: + def refuse(*_args, **_kwargs): + raise AssertionError("the hook process opened a database") + + monkeypatch.setattr(sqlite3, "connect", refuse) + monkeypatch.setattr("one_memory.database_access.main_database", refuse) + + +@pytest.mark.parametrize("receipt", ["missing", "dead_pid", "nobody_listening"]) +def test_with_core_unreachable_capture_drops_the_turn_fast_and_writes_nothing( + tmp_path, monkeypatch, receipt +) -> None: + data_dir = tmp_path / "data" + data_dir.mkdir() + if receipt == "dead_pid": + _receipt(data_dir, port=_free_port(), pid=2**22 + 1) + if receipt == "nobody_listening": + _receipt(data_dir, port=_free_port(), pid=os.getpid()) + _no_database(monkeypatch) + prompt = {"hook_event_name": "UserPromptSubmit", "session_id": "s", "prompt": "记得用深绿色"} + + started = time.monotonic() + assert _hook(monkeypatch, data_dir, "workbuddy", prompt) == (0, "") + + assert time.monotonic() - started < capture_handoff.HANDOFF_TIMEOUT_SECONDS + 0.5 + assert sorted(path.name for path in data_dir.iterdir()) == sorted( + [] if receipt == "missing" else ["runtime.json", "runtime-token"] + ) + + +def test_a_slow_core_is_given_up_on_within_the_hand_off_timeout(tmp_path, monkeypatch) -> None: + data_dir = tmp_path / "data" + listener = socket.socket() + listener.bind(("127.0.0.1", 0)) + listener.listen(1) + try: + # Accepts the connection and never answers: a Core stuck behind its writer. + _receipt(data_dir, port=listener.getsockname()[1], pid=os.getpid()) + _no_database(monkeypatch) + prompt = {"hook_event_name": "UserPromptSubmit", "session_id": "s", "prompt": "蓝色"} + started = time.monotonic() + assert _hook(monkeypatch, data_dir, "workbuddy", prompt) == (0, "") + assert time.monotonic() - started < capture_handoff.HANDOFF_TIMEOUT_SECONDS + 1.0 + finally: + listener.close() + + +def test_capture_is_a_silent_noop_during_sdk_migration(tmp_path, monkeypatch) -> None: + data_dir = tmp_path / "data" + data_dir.mkdir() + marker = data_dir / ".sdk-migration-capture-fence.json" + marker.write_text(json.dumps({"schema_version": 1, "pid": os.getpid()})) + + def handed(*_args, **_kwargs): + raise AssertionError("a fenced capture reached Core") + + monkeypatch.setattr(capture_handoff, "hand_to_core", handed) + payload = {"hook_event_name": "beforeSubmitPrompt", "conversation_id": "c", "prompt": "x"} + + assert _hook(monkeypatch, data_dir, "cursor", payload) == (0, "") + assert not (data_dir / "one-memory.sqlite3").exists() + + +def test_a_bad_payload_or_a_failing_hand_off_stays_silent(tmp_path, monkeypatch) -> None: + data_dir = tmp_path / "data" + data_dir.mkdir() + + def explode(*_args, **_kwargs): + raise RuntimeError("socket gone") + + monkeypatch.setattr(capture_handoff, "hand_to_core", explode) + assert _hook(monkeypatch, data_dir, "workbuddy", {"hook_event_name": "Stop"}) == (0, "") + assert _hook(monkeypatch, data_dir, "workbuddy", ["not", "an", "object"]) == (0, "") + monkeypatch.setattr("sys.stdin", StringIO("{not json")) + assert cli.run(["capture", "workbuddy"], output=StringIO()) == 0 + assert cli.run(["capture", "elsewhere"], output=StringIO()) == 2 diff --git a/memhub/apps/core/tests/test_sdk_migration_restore.py b/memhub/apps/core/tests/test_sdk_migration_restore.py index 5ef4845..f5f16f7 100644 --- a/memhub/apps/core/tests/test_sdk_migration_restore.py +++ b/memhub/apps/core/tests/test_sdk_migration_restore.py @@ -293,6 +293,22 @@ def _retry_succeeds(home: Path, monkeypatch) -> None: assert len(list((home / "backups").glob("*-pre-cutover.sdk-bundle"))) == 1 +def test_evidence_an_older_capture_hook_recreated_is_dropped_on_the_next_start(home): + # Builds through 1.0.6 let the hook process open `EvidenceStore`, which + # rebuilt the retired tables on an SDK home. Nothing reads them there, and + # the next startup migration's leftovers cleanup removes them again. + assert _run(home).state == "cutover" + database = home / "one-memory.sqlite3" + EvidenceStore(database) + assert "evidence" in legacy_tables_present(database) + + result = _run(home) + + assert result.state == "complete" + assert {"evidence", "evidence_search"} <= set(result.dropped_tables) + assert legacy_tables_present(database) == () + + def test_a_failure_after_the_cutover_commit_is_not_restored(home, monkeypatch): import one_memory.sdk_migration_runner as runner diff --git a/memhub/apps/core/tests/test_service_level_pressure_acceptance.py b/memhub/apps/core/tests/test_service_level_pressure_acceptance.py index b56d0df..df0b2a6 100644 --- a/memhub/apps/core/tests/test_service_level_pressure_acceptance.py +++ b/memhub/apps/core/tests/test_service_level_pressure_acceptance.py @@ -23,9 +23,8 @@ def test_workbuddy_recall_remains_available_during_100k_import(tmp_path, monkeyp assert result["import_status"] == "completed" assert result["stored"] == 100_000 - assert result["evidence_rows"] == 100_001 + assert result["evidence_rows"] == 100_000 assert result["hook_accepted"] is True - assert result["hook_stored"] is True assert result["hook_visible_ms"] < 10_000 assert result["recall_sampling_terminal"] is True assert result["during_unavailable"] == 0 diff --git a/memhub/docs/ARCHITECTURE.md b/memhub/docs/ARCHITECTURE.md index ad233fd..e1b3483 100644 --- a/memhub/docs/ARCHITECTURE.md +++ b/memhub/docs/ARCHITECTURE.md @@ -139,6 +139,14 @@ This exception does not extend to VDB, Coding, prompt, reconcile, or lifecycle. - 旧用户迁移：由 App 在可见的启动 loading 阶段统一执行；此时 Core 尚未对外提供产品读写。所有已发布版本都先经 SDK 写入，Governance/Evidence 只是镜像，所以迁移**不复制任何 memory 或 raw**：创建并校验完整 bundle、让写入队列脱离旧表、把旧版命名的 SDK 向量集合接回当前名字、用 Runtime 同样配置的公开 SDK client 按 ID 回读核对，通过后才提交 marker、删除旧表。由一个 Core 命令 `sdk-migrate` 完成；Desktop 启动与 CLI `memhub install`/`upgrade`（macOS、Linux）在启动新 Runtime 前调用它，普通用户不需要另跑命令。App 不得在 migration `deferred` 状态继续启动 Runtime。 - 迁移期间宿主 capture hook 由 Desktop 活跃 PID marker 静默挡住，不再追加 Evidence；cutover 后先从宿主 History checkpoint 补齐遗漏 turn，再恢复 live sync。 +- 宿主 capture hook（`one-memory capture `）是短命的独立进程，**不打开主库**： + 它只读 runtime receipt 与 runtime token，把 hook JSON 经 loopback + `POST /api/v1/live-capture/{source}` 交给正在运行的 Core，超时或 Core 未运行即静默丢弃， + 不回退本地写；迁移 fence 仍在 hook 侧检查。Core 在进程内套用其余的门（grant、sync、 + 拒绝计数、自动化会话、cursor/claude 去重），再按 live file sync 的同一路径 + （`materialize_write_input` → `stage_sdk_input` → `finalize_sdk_file`，live 优先级） + 经单一 writer 入队。live sync 已读取其宿主文件的来源，hook 内容交给文件路径， + 同一 turn 只进 SDK 一次；只有没有文件轮询的来源才从 hook 入队。 - Matter 若未来重新开发，需以新的 SDK-native 合同重新设计，不继承当前占位 schema。 Pinned SDK 版本见发布锁文件 / `MEMORY.md`（随发布更新）。 diff --git a/memhub/packages/host-bridge/src/capture-hooks.ts b/memhub/packages/host-bridge/src/capture-hooks.ts index 54a04c5..c87c7b3 100644 --- a/memhub/packages/host-bridge/src/capture-hooks.ts +++ b/memhub/packages/host-bridge/src/capture-hooks.ts @@ -1,8 +1,9 @@ /** Register `capture` on the hosts whose one-click install can own a hook file. * * The command is the packaged core binary. It reads the hook's JSON on stdin - * and writes visible text into Evidence. Stdout stays empty, so the host does - * not treat the capture as another instruction. + * and hands it to the running Core over loopback; it never opens the + * database, and drops the turn when Core is not running. Stdout stays empty, + * so the host does not treat the capture as another instruction. * * WorkBuddy's command lives in its plugin tree. Hermes and OpenClaw deliver * the turn through a single memory-provider slot; those are not written here,