Merge pull request #344 from Tencent-Hunyuan/fix/memhub-hooks-no-evidence

fix(memhub): capture hooks hand turns to Core instead of writing Evidence

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 <host>` 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

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,

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 <source>` 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

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"]):

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:

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

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. |

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(

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()

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
 

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

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 <host>`）是短命的独立进程，**不打开主库**：
+  它只读 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`（随发布更新）。

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,