chore(release): bump MemHub build number to 33

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

memhub/apps/core/src/one_memory/api.py

@@ -47,8 +47,8 @@ from one_memory.background_work import shutdown_bulk_pool, warm_bulk_pool
 from one_memory.config import Settings, load_settings
 from one_memory.context_routes import sdk_context_router
 from one_memory.conversation_processing import (
-    _conversation_chunks,
     memory_task_timeout_seconds,
+    turn_batches,
 )
 from one_memory.database_access import (
     MAIN_DATABASE_REPORT_SECONDS,
@@ -95,6 +95,8 @@ from one_memory.hy_sdk_runtime import (
     SdkClientHandle,
     builtin_retrieval_snapshot,
     close_all_shared_runtimes,
+    hold_stores_offline,
+    release_stores,
     sdk_relevance_floor,
 )
 from one_memory.hy_sdk_types import (
@@ -223,6 +225,7 @@ from one_memory.sdk_memory_qa import (
     SdkMemoryAnswerer,
 )
 from one_memory.sdk_memory_runtime import SdkMemoryRuntime
+from one_memory.sdk_migration_log import error_chain_text, error_frames
 from one_memory.sdk_write_processor import SdkWriteCoordinator
 from one_memory.secrets import (
     KeyringSecretStore,
@@ -710,7 +713,8 @@ def create_app(
                 raise SdkMigrationRequired()
             if app.state.memory_authority is MemoryAuthority.SDK:
                 # The metadata repair after a cutover reads every bundle: it is
-                # recovery work, so it runs after readiness (`recover_after_ready`).
+                # recovery work, so it runs after readiness (`recover_after_ready`)
+                # while the SDK's stores are held offline (see there).
                 app.state.memory_governance_store = RetiredGovernanceStore(database_path)
                 app.state.evidence_store = RetiredEvidenceStore(database_path)
             else:
@@ -1026,28 +1030,25 @@ def create_app(
 
             app.state.memory_write_intake = UnifiedWriteIntake(
                 memory_write_queue,
-                source_writes_permitted=lambda source_id: (
-                    source_id == "one-memory-manual" or _source_writes_on(app, source_id)
-                ),
+                source_writes_permitted=lambda source_id: _source_writes_on(app, source_id),
             )
             app.state.start_memory_backfill = lambda: _start_backfill(app)
 
             async def enqueue_stable_evidence(record: EvidenceRecord) -> None:
-                priority = "live" if live_sync_arrival.get() else "normal"
+                live = live_sync_arrival.get()
                 conversation = await asyncio.to_thread(
                     app.state.evidence_store.list_conversation, record.evidence_id
                 )
-                for chunk in _conversation_chunks(conversation, max_turns=12, completed_only=True):
-                    if (
-                        record not in chunk
-                        or not any(item.role == "user" for item in chunk)
-                        or not any(item.role == "assistant" for item in chunk)
-                    ):
-                        continue
+                # Legacy authority. The turn rule groups the whole conversation
+                # from its start, so a batch's records and key never change:
+                # history (end of file) writes it all, live writes complete
+                # batches and leaves the rest to catch-up (`_run_backfill`).
+                batches = turn_batches(conversation, flush=not live).batches
+                for batch in _batches_settled_by(batches, record):
                     await asyncio.to_thread(
                         app.state.memory_write_intake.submit_records,
-                        chunk,
-                        priority=priority,
+                        batch,
+                        priority="live" if live else "normal",
                     )
 
             app.state.memory_addition_store = MemoryAdditionStore(
@@ -1189,10 +1190,14 @@ def create_app(
                     except Exception as error:
                         failed.append(name)
                         app.state.recovery_failed_steps = tuple(failed)
+                        # Where it failed, as `file:line:function` frames of
+                        # the exception chain, never its message: a bare
+                        # `error_type=KeyError` named no location.
                         logger.warning(
-                            "core_recovery_step_failed step=%s error_type=%s",
+                            "core_recovery_step_failed step=%s error_type=%s chain=%s",
                             name,
                             type(error).__name__,
+                            error_chain_text(error),
                         )
                         diagnostic_event(
                             level="warn",
@@ -1200,20 +1205,42 @@ def create_app(
                             event="core.recovery_step_failed",
                             outcome="failed",
                             error_type=type(error).__name__,
+                            safe_stack=error_frames(error) or None,
                             duration_ms=elapsed_ms(step_started),
                             details={"phase": name},
                         )
 
                 async def repair_sdk_metadata() -> None:
+                    # `sdk-migrate` repairs right after its cutover and writes
+                    # the receipt, so this normally finds it and returns. It
+                    # stays for homes whose repair failed before that (no
+                    # receipt). It opens its own Chroma client on the SDK's
+                    # vector store, and Chroma's shared System is not safe
+                    # beside another client: a first library request that
+                    # opened the SDK client meanwhile lost it (KeyError).
+                    # The stores were held offline before readiness, so any
+                    # SDK client waits for this step.
                     from one_memory.hy_sdk_migration import (
                         repair_sdk_metadata_from_bundle,
                     )
 
-                    repair = await asyncio.to_thread(
-                        repair_sdk_metadata_from_bundle,
-                        data_dir=runtime_settings.data_dir,
-                        database_path=database_path,
+                    if not app.state.sdk_stores_held:
+                        logger.warning("sdk_metadata_repair_deferred reason=sdk_client_open")
+                        return
+                    # The hold is this step's from here; shutdown leaves it.
+                    app.state.sdk_stores_held = False
+                    work = asyncio.ensure_future(
+                        asyncio.to_thread(
+                            repair_sdk_metadata_from_bundle,
+                            data_dir=runtime_settings.data_dir,
+                            database_path=database_path,
+                        )
                     )
+                    # Released when the thread ends, not when this coroutine
+                    # is cancelled: a cancelled await leaves the thread
+                    # writing Chroma.
+                    work.add_done_callback(lambda _work: release_stores(runtime_settings.data_dir))
+                    repair = await asyncio.shield(work)
                     if repair.vector_rows or repair.coding_rows:
                         logger.info(
                             "sdk_metadata_repair_completed mapped=%d vector=%d coding=%d",
@@ -1337,7 +1364,12 @@ def create_app(
                 )
 
             app.state.recovery_task = None
+            app.state.sdk_stores_held = False
             try:
+                if app.state.memory_authority is MemoryAuthority.SDK:
+                    # Before the first request can open an SDK client; the
+                    # metadata repair step releases it.
+                    app.state.sdk_stores_held = hold_stores_offline(runtime_settings.data_dir)
                 diagnostic_event(
                     level="info",
                     component="core.lifecycle",
@@ -1394,6 +1426,10 @@ def create_app(
                 if app.state.live_sync_watch is not None:
                     await app.state.live_sync_watch.shutdown()
                 await app.state.live_sync_supervisor.shutdown()
+                if app.state.sdk_stores_held:
+                    # Recovery never reached the metadata repair step.
+                    release_stores(runtime_settings.data_dir)
+                    app.state.sdk_stores_held = False
                 await app.state.import_supervisor.shutdown()
                 app.state.memory_write_drain.stop()
                 if app.state.memory_write_drain_task is not None:
@@ -2824,11 +2860,24 @@ def create_app(
         dependencies=protected,
     )
     async def sdk_home_summary() -> SdkHomeSummary:
+        """The home card: the memory page's count and its newest memory.
+
+        The count is `memory_stats` total (chat + working + coding
+        experiences, facts not counted), the same set the memory page lists
+        with scene="all". Without a scene `list_memories` counts chat only,
+        which showed 55 on the home card against 94 on the memory page.
+        """
         try:
             client = await _sdk_client_handle(app).client()
+            stats = await asyncio.to_thread(
+                client.memory_stats,
+                user_id="local",
+                statuses=["active"],
+            )
             raw = await asyncio.to_thread(
                 client.list_memories,
                 user_id="local",
+                scene="all",
                 statuses=["active"],
                 limit=1,
                 offset=0,
@@ -2838,8 +2887,9 @@ def create_app(
             _log_skipped_sdk_memories(skipped)
         except Exception as error:
             raise _sdk_http_error(error, "hy_memory_list_failed") from None
+        total = stats.get("total") if isinstance(stats, dict) else None
         return SdkHomeSummary(
-            memory_count=page.total,
+            memory_count=total if isinstance(total, int) and total >= 0 else page.total,
             latest=page.memories[0] if page.memories else None,
             observed_at=datetime.now(UTC),
         )
@@ -4330,14 +4380,11 @@ async def _run_backfill(app: FastAPI) -> None:
                     _evidence_store(app).list_conversation,
                     evidence_id,
                 )
-                for chunk in _conversation_chunks(
-                    conversation,
-                    max_turns=12,
-                    completed_only=True,
-                ):
-                    if not any(item.evidence_id in evidence_ids for item in chunk):
+                # Catch-up is history: the last turn and a short batch are written.
+                for batch in turn_batches(conversation, flush=True).batches:
+                    if not any(item.evidence_id in evidence_ids for item in batch):
                         continue
-                    await asyncio.to_thread(intake.submit_records, chunk, priority="normal")
+                    await asyncio.to_thread(intake.submit_records, batch, priority="normal")
             cursor = evidence_ids[-1]
     except asyncio.CancelledError:
         raise
@@ -4356,6 +4403,22 @@ async def _run_backfill(app: FastAPI) -> None:
         logger.exception("memory catch-up failed, no further memories will be made")
 
 
+def _batches_settled_by(
+    batches: tuple[tuple[EvidenceRecord, ...], ...],
+    record: EvidenceRecord,
+) -> list[tuple[EvidenceRecord, ...]]:
+    """The batches one arriving record can have made writable.
+
+    Its own batch, and the last two: a user message completes the turn
+    before it, which can close that turn's batch and, at the 8000-character
+    cut, the batch before it. Submitting is idempotent, so an older batch
+    named again is only a hit.
+    """
+    chosen = [batch for batch in batches if record in batch]
+    chosen.extend(batch for batch in batches[-2:] if batch not in chosen)
+    return chosen
+
+
 def _sdk_memory_runtime(app: FastAPI) -> SdkMemoryRuntime:
     return cast(
         SdkMemoryRuntime,
@@ -4651,10 +4714,17 @@ def _evidence_store(app: FastAPI) -> EvidenceStore:
 
 
 def _source_writes_on(app: FastAPI, source_id: str) -> bool:
-    """Whether this source's conversations may become memories."""
+    """Whether this source's conversations may become memories.
+
+    A hand-added memory is allowed unless its grant was explicitly revoked;
+    a missing grant row is allowed. The SDK add path writes no grant row, so
+    requiring one left every SDK-authority manual add unclaimable: Pending
+    forever with attempts=0 while later jobs ran. Intake, claim and execute
+    all ask this one question.
+    """
     if source_id == "one-memory-manual":
         receipt = _grant_store(app).receipt(user_id="local", source_id=source_id)
-        return receipt is not None and receipt.revoked_at is None
+        return receipt is None or receipt.revoked_at is None
     return _source_processing_enabled(app, source_id)
 
 

memhub/apps/core/src/one_memory/conversation_processing.py

@@ -4,9 +4,10 @@ from __future__ import annotations
 
 import os
 import re
+from collections.abc import Sequence
+from dataclasses import dataclass
 
-from one_memory.imports.models import EvidenceContentKind, EvidenceOrigin, EvidenceRecord
-from one_memory.memory_tasks import segment_evidence_record
+from one_memory.imports.models import EvidenceContentKind, EvidenceRecord
 
 MEMORY_TASK_TIMEOUT_MARGIN_SECONDS = 60.0
 MEMORY_TASK_TIMEOUT_MIN_SECONDS = 180.0
@@ -29,10 +30,13 @@ _HOST_ENVELOPE = re.compile(
     r")",
     re.IGNORECASE | re.DOTALL,
 )
-_WHOLE_TURN_CHARACTERS = 60_000
-_WINDOW_BYTE_CEILING = 2_000_000
-_HISTORY_MERGE_TURNS = 8
-_HISTORY_MERGE_CHARACTERS = 8_000
+#: The write rule (`turn_batches`): every 5 complete turns, or fewer when
+#: the next turn would take the batch past 8000 HY-read characters.
+TURN_BATCH_TURNS = 5
+TURN_BATCH_CHARACTERS = 8_000
+#: A live conversation's last turn counts as complete, and a short batch is
+#: written, after this long without a new message in that conversation.
+TURN_IDLE_FLUSH_SECONDS = 600
 
 
 def memory_task_timeout_seconds(model_timeout_seconds: float) -> float:
@@ -57,180 +61,83 @@ def _coding_tool_result_max_bytes() -> int:
 _HY_TOOL_CHARACTERS = _coding_tool_result_max_bytes()
 
 
-def _conversation_chunks(
-    records: list[EvidenceRecord],
-    *,
-    max_characters: int | None = None,
-    max_turns: int | None = None,
-    completed_only: bool = False,
-) -> list[tuple[EvidenceRecord, ...]]:
-    """Build bounded, complete user/assistant/tool windows."""
+@dataclass(frozen=True)
+class TurnBatches:
+    """What a conversation's records become: batches to write, records to keep."""
+
+    #: Each batch is whole turns; every batch holds at least one user message.
+    batches: tuple[tuple[EvidenceRecord, ...], ...]
+    #: Not written yet: the open last turn, complete turns short of a batch,
+    #: or records with no user message before them. Fed back in, in order,
+    #: with the conversation's next records.
+    carry: tuple[EvidenceRecord, ...]
+
+
+def conversation_turns(records: Sequence[EvidenceRecord]) -> list[tuple[EvidenceRecord, ...]]:
+    """One user message and every record after it up to the next user message.
+
+    Records before the first user message are not a turn of their own: they
+    join the first turn (an assistant proposal and the user turn adopting it
+    reach HY together). With no user message at all the result is one group
+    that does not open with a user, which `turn_batches` never writes.
+    """
     turns: list[list[EvidenceRecord]] = []
     for record in records:
         if not _is_hy_turn_record(record):
             continue
-        pending_proposal = bool(turns) and not _opens_turn(turns[-1][0])
-        if not turns or (_opens_turn(record) and not pending_proposal):
+        # Until the first user message, the group is leading records only.
+        leading = bool(turns) and not any(_opens_turn(item) for item in turns[-1])
+        if not turns or (_opens_turn(record) and not leading):
             turns.append([record])
         else:
             turns[-1].append(record)
-
-    if completed_only and turns and _trailing_user_turn_is_open(turns[-1]):
-        turns.pop()
-
-    whole = _whole_turns() and max_characters is None
-    if whole:
-        size = _hy_read_length
-        max_characters = _WHOLE_TURN_CHARACTERS
-        max_turns = max_turns or 1_000_000
-    else:
-        size = lambda record: len(record.verbatim)  # noqa: E731
-        max_characters = max_characters or 12_000
-        max_turns = max_turns or 24
-
-    chunks: list[tuple[EvidenceRecord, ...]] = []
-    for turn in turns:
-        original_question = turn[0] if _opens_turn(turn[0]) else None
-        expanded: list[EvidenceRecord] = []
-        for record in turn:
-            reserve = (
-                size(original_question)
-                if original_question is not None
-                and record is not original_question
-                and size(original_question) < max_characters
-                else 0
-            )
-            segment_limit = max(1, max_characters - reserve)
-            if size(record) <= segment_limit:
-                expanded.append(record)
-                continue
-            spans = _semantic_segment_spans(record.verbatim, segment_limit)
-            expanded.extend(
-                segment_evidence_record(record, index=index, start=start, end=end)
-                if len(spans) > 1
-                else record
-                for index, (start, end) in enumerate(spans)
-            )
-        question = expanded[0] if expanded and _opens_turn(expanded[0]) else None
-        current: list[EvidenceRecord] = []
-        for record in expanded:
-            if current and (
-                len(current) >= max_turns
-                or sum(size(item) for item in current) + size(record) > max_characters
-                or sum(_window_bytes(item) for item in current) + _window_bytes(record)
-                > _WINDOW_BYTE_CEILING
-            ):
-                chunks.append(tuple(current))
-                current = (
-                    [question]
-                    if question is not None
-                    and max_turns > 1
-                    and size(question) + size(record) <= max_characters
-                    else []
-                )
-            current.append(record)
-        chunks.append(tuple(current))
-    return _merge_tool_only_chunks(chunks)
+    return [tuple(turn) for turn in turns]
 
 
-def _semantic_segment_spans(text: str, limit: int) -> list[tuple[int, int]]:
-    if len(text) <= limit:
-        return [(0, len(text))]
-    spans: list[tuple[int, int]] = []
-    start = 0
-    separators = ("\n\n", "\n", "。", "！", "？", ". ", "! ", "? ")
-    while start < len(text):
-        hard_end = min(len(text), start + limit)
-        if hard_end == len(text):
-            end = hard_end
-        else:
-            floor = start + max(1, limit // 2)
-            candidates = [text.rfind(separator, floor, hard_end) for separator in separators]
-            boundary = max(candidates)
-            if boundary < floor:
-                end = hard_end
-            else:
-                matched = next(
-                    separator for separator in separators if text.startswith(separator, boundary)
-                )
-                end = boundary + len(matched)
-        spans.append((start, end))
-        start = end
-    return spans
+def turn_batches(records: Sequence[EvidenceRecord], *, flush: bool) -> TurnBatches:
+    """Group a conversation's records into write batches by the turn rule.
 
-
-def _merge_tool_only_chunks(
-    chunks: list[tuple[EvidenceRecord, ...]],
-) -> list[tuple[EvidenceRecord, ...]]:
-    if not any(chunk and all(record.role == "tool" for record in chunk) for chunk in chunks):
-        return chunks
-    merged: list[tuple[EvidenceRecord, ...]] = []
-    leading_tools: list[EvidenceRecord] = []
-    for chunk in chunks:
-        if not chunk:
-            continue
-        if all(record.role == "tool" for record in chunk):
-            if merged:
-                merged[-1] = (*merged[-1], *chunk)
-            else:
-                leading_tools.extend(chunk)
-            continue
-        if leading_tools:
-            chunk = (*leading_tools, *chunk)
-            leading_tools = []
-        merged.append(chunk)
-    return [] if leading_tools else merged
-
-
-def _merge_historical_chunks(
-    chunks: list[tuple[EvidenceRecord, ...]],
-    *,
-    max_turns: int = _HISTORY_MERGE_TURNS,
-    max_characters: int = _HISTORY_MERGE_CHARACTERS,
-) -> list[tuple[EvidenceRecord, ...]]:
-    if not chunks:
-        return []
-    merged: list[tuple[EvidenceRecord, ...]] = []
-    current: list[EvidenceRecord] = []
-    current_turns = 0
+    - A turn is complete once the next user message follows it; `flush`
+      completes the last one too (history at end of file, or 10 idle minutes).
+    - Every 5 complete turns are one batch. Before 5, a turn that would take
+      the batch past 8000 characters starts the next batch; one turn over
+      8000 is a batch by itself and is never split.
+    - Characters are what HY reads: user/assistant text in full, a tool
+      record up to `_HY_TOOL_CHARACTERS`.
+    - Fewer than 5 complete turns within 8000 are written only on `flush`.
+    - No batch is written without a user message; records with none before
+      them are carried, and dropped on `flush`.
+    """
+    turns = conversation_turns(records)
+    if turns and not any(_opens_turn(record) for record in turns[0]):
+        # Records with no user message yet: nothing to write until one comes.
+        return TurnBatches(batches=(), carry=() if flush else turns[0])
+    complete = turns if flush else turns[:-1]
+    open_tail = () if flush or not turns else turns[-1]
+    batches: list[tuple[EvidenceRecord, ...]] = []
+    current: list[tuple[EvidenceRecord, ...]] = []
     current_characters = 0
-    for chunk in chunks:
-        chunk_characters = sum(len(record.verbatim) for record in chunk)
-        if current and (
-            current_turns >= max_turns or current_characters + chunk_characters > max_characters
-        ):
-            merged.append(tuple(current))
-            current = []
-            current_turns = 0
-            current_characters = 0
-        current.extend(chunk)
-        current_turns += 1
-        current_characters += chunk_characters
-    if current:
-        merged.append(tuple(current))
-    return merged
-
-
-def _chunk_input_ref(chunk: tuple[EvidenceRecord, ...]) -> str:
-    user_records = tuple(item for item in chunk if item.role == "user")
-    return user_records[-1].evidence_id if user_records else chunk[-1].evidence_id
+    for turn in complete:
+        characters = sum(_hy_read_length(record) for record in turn)
+        if current and current_characters + characters > TURN_BATCH_CHARACTERS:
+            batches.append(tuple(record for item in current for record in item))
+            current, current_characters = [], 0
+        current.append(turn)
+        current_characters += characters
+        if len(current) >= TURN_BATCH_TURNS or current_characters > TURN_BATCH_CHARACTERS:
+            batches.append(tuple(record for item in current for record in item))
+            current, current_characters = [], 0
+    if current and flush:
+        batches.append(tuple(record for item in current for record in item))
+        current = []
+    carry = tuple(record for item in current for record in item) + tuple(open_tail)
+    return TurnBatches(batches=tuple(batches), carry=carry)
 
 
 def _opens_turn(record: EvidenceRecord) -> bool:
     return record.role == "user"
 
 
-def _trailing_user_turn_is_open(turn: list[EvidenceRecord]) -> bool:
-    if len(turn) != 1:
-        return False
-    opener = turn[0]
-    return (
-        _opens_turn(opener)
-        and opener.content_kind is EvidenceContentKind.RAW_CONVERSATION
-        and not _is_hand_added_memory(opener)
-    )
-
-
 def _is_hy_turn_record(record: EvidenceRecord) -> bool:
     if record.role == "tool":
         return record.content_kind is EvidenceContentKind.RAW_CONVERSATION and bool(
@@ -254,23 +161,6 @@ def _is_hy_input_turn(record: EvidenceRecord) -> bool:
     return _HOST_ENVELOPE.search(source[:2_000]) is None
 
 
-def _is_hand_added_memory(record: EvidenceRecord) -> bool:
-    meta = record.origin_meta.get(EvidenceOrigin.MANUAL_IMPORT)
-    return (
-        record.source_id == "one-memory-manual"
-        and meta is not None
-        and meta.raw_meta.get("input_mode") == "direct_memory_note"
-    )
-
-
-def _window_bytes(record: EvidenceRecord) -> int:
-    return len(record.verbatim.encode("utf-8"))
-
-
-def _whole_turns() -> bool:
-    return os.environ.get("ONE_MEMORY_SPLIT_TURNS", "").strip() != "1"
-
-
 def _hy_read_length(record: EvidenceRecord) -> int:
     if record.role == "tool":
         return min(len(record.verbatim), _HY_TOOL_CHARACTERS)

memhub/apps/core/src/one_memory/hy_sdk_migration.py

@@ -82,7 +82,10 @@ def repair_sdk_metadata_from_bundle(
     database_path: Path,
     now: datetime | None = None,
 ) -> MetadataRepairResult:
-    """Restore metadata lost by historical cutovers from their verified bundles."""
+    """Restore metadata lost by historical cutovers from their verified bundles.
+
+    Offline only: no SDK client may be open on this home (see `_repair_vector`).
+    """
     migration_id = _sdk_migration_id(database_path)
     if migration_id is None:
         return MetadataRepairResult(None, 0, 0, 0)
@@ -329,8 +332,11 @@ def _repair_vector(
     from one_memory.sdk_vector_reattach import open_offline_chroma
 
     client = open_offline_chroma(directory)
-    # The SDK client is not open yet (startup has not created its handle), so
-    # this is the only writer; release the store before the SDK takes it.
+    # This must be the only Chroma client on the directory: Chroma shares one
+    # System per path, and this client's close() drops it from under any
+    # other (KeyError on its next lookup). `sdk-migrate` runs it after the
+    # cutover with Runtime stopped; Core's startup fallback runs it while
+    # `hold_stores_offline` keeps every SDK client waiting.
     try:
         repaired = 0
         ids = tuple(mapping)

memhub/apps/core/src/one_memory/hy_sdk_runtime.py

@@ -36,7 +36,11 @@ from one_memory.hy_graph_admin import (
 )
 from one_memory.hy_sdk_types import HySdkRuntimeError
 from one_memory.local_embedding import LocalEmbedder
-from one_memory.model_profiles import ModelProfileSnapshot, ProviderKind
+from one_memory.model_profiles import (
+    ModelProfileSnapshot,
+    ProviderKind,
+    disable_thinking_params,
+)
 
 _RUNTIME_LOCK = threading.RLock()
 BUILTIN_RETRIEVAL_PROFILE_ID = "builtin-retrieval"
@@ -52,6 +56,16 @@ class _RuntimeEntry:
 
 _RUNTIMES: dict[Path, _RuntimeEntry] = {}
 
+# Homes whose SDK-owned local stores are held offline for maintenance, and
+# the client constructions in flight per home. Chroma keeps one process-wide
+# System per path with no lock around create, lookup and release, so a
+# second client that opens or closes beside the SDK's pulls the System out
+# from under it (KeyError on the next lookup). Maintenance therefore runs
+# only while no SDK client exists for that home, and new clients wait.
+_STORE_GATE = threading.Condition()
+_OFFLINE_STORES: set[Path] = set()
+_OPENING: dict[Path, int] = {}
+
 
 class LocalOnnxEmbeddingService:
     """Public ``EmbeddingService`` implementation backed by MemHub's ONNX model."""
@@ -244,14 +258,22 @@ def build_sdk_config(
         embedding_provider, embedding_endpoint, embedding_model, dimensions
     )
 
+    llm: dict[str, Any] = {
+        "provider": "openai",
+        "model": profile.model,
+        "api_key": generation_key,
+        "base_url": generation_endpoint,
+    }
+    # The SDK sends `extra_body` on every chat call, chat and coding/working
+    # alike. Without it DeepSeek spent the whole agent_max_tokens on thinking
+    # and answered finish_reason=length with no content.
+    thinking_off = disable_thinking_params(profile)
+    if thinking_off:
+        llm["extra_body"] = thinking_off
+
     return MemoryConfig.from_dict(
         {
-            "llm": {
-                "provider": "openai",
-                "model": profile.model,
-                "api_key": generation_key,
-                "base_url": generation_endpoint,
-            },
+            "llm": llm,
             "embedder": {
                 "provider": "openai",
                 "model": (
@@ -295,7 +317,60 @@ def acquire_client(
     data_dir: Path,
     http_transport_factory: Callable[[], Any] | None = None,
 ) -> ClientLease:
-    """Construct a public SDK client and return its explicit runtime lease."""
+    """Construct a public SDK client and return its explicit runtime lease.
+
+    Waits while `hold_stores_offline` holds this home's stores.
+    """
+    key = data_dir.resolve()
+    with _STORE_GATE:
+        while key in _OFFLINE_STORES:
+            _STORE_GATE.wait()
+        _OPENING[key] = _OPENING.get(key, 0) + 1
+    try:
+        return _acquire_client(
+            profile,
+            api_key=api_key,
+            data_dir=data_dir,
+            http_transport_factory=http_transport_factory,
+        )
+    finally:
+        with _STORE_GATE:
+            _OPENING[key] -= 1
+            if not _OPENING[key]:
+                del _OPENING[key]
+            _STORE_GATE.notify_all()
+
+
+def hold_stores_offline(data_dir: Path) -> bool:
+    """Hold this home's SDK stores for offline maintenance; False if a client is open.
+
+    While held, `acquire_client` for the home waits. Every SDK client in Core
+    is built there, so nothing reads or writes Chroma beside the maintenance.
+    `release_stores` ends the hold; it is idempotent.
+    """
+    key = data_dir.resolve()
+    with _STORE_GATE:
+        with _RUNTIME_LOCK:
+            open_runtime = key in _RUNTIMES
+        if key in _OFFLINE_STORES or _OPENING.get(key) or open_runtime:
+            return False
+        _OFFLINE_STORES.add(key)
+        return True
+
+
+def release_stores(data_dir: Path) -> None:
+    with _STORE_GATE:
+        _OFFLINE_STORES.discard(data_dir.resolve())
+        _STORE_GATE.notify_all()
+
+
+def _acquire_client(
+    profile: ModelProfileSnapshot,
+    *,
+    api_key: str | None,
+    data_dir: Path,
+    http_transport_factory: Callable[[], Any] | None,
+) -> ClientLease:
     config = build_sdk_config(profile, api_key=api_key, data_dir=data_dir)
     config.http_transport_factory = http_transport_factory
     mismatch = inspect_graph_width_mismatch(config)

memhub/apps/core/src/one_memory/imports/runner.py

@@ -662,11 +662,15 @@ class ImportRunner:
                         and self.sdk_queue is not None
                         and event.cursor.read_to_end
                     ):
+                        # History ends at this end of file, so its last turn
+                        # and a short batch are written. Live polling keeps
+                        # them for the next increment or the idle flush.
                         self.sdk_queue.finalize_sdk_file(
                             user_id=current.user_id,
                             source_id=current.source_id,
                             file_key=event.cursor.file_key,
                             connection=connection,
+                            flush=not self._live_import,
                         )
                         continue
                     if self.on_stable_evidence is not None and event.cursor.read_to_end:

memhub/apps/core/src/one_memory/live_capture.py

@@ -256,24 +256,19 @@ def record_capture_rejection(
 
 
 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.conversation_processing import _conversation_chunks
     from one_memory.memory_server.durable_queue import DurableWriteQueue
-    from one_memory.memory_server.unified_entry import UnifiedWriteIntake
 
-    intake = UnifiedWriteIntake(DurableWriteQueue(store.database_path, evidence=store))
-    for chunk in _conversation_chunks(
-        store.list_conversation(record.evidence_id),
-        max_turns=12,
-        completed_only=True,
-    ):
-        if (
-            record in chunk
-            and any(item.role == "user" for item in chunk)
-            and any(item.role == "assistant" for item in chunk)
-        ):
-            intake.submit_records(chunk, priority="live")
+    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:

memhub/apps/core/src/one_memory/memory_additions.py

@@ -130,10 +130,18 @@ class MemoryAdditionStore:
                 raise RuntimeError("sdk_memory_add_intake_unavailable")
             record = materialize_write_input(value, observed_at=now)
             with main_database(self.write_queue.database_path).read() as database:
+                # No grant row is written here: a missing row means allowed,
+                # only an explicit revocation refuses (same as the claim gate).
+                grant = database.execute(
+                    "SELECT revoked_at FROM source_grants WHERE user_id='local' AND source_id=?",
+                    (SOURCE_ID,),
+                ).fetchone()
                 existing = database.execute(
                     "SELECT request_id FROM memory_note_requests WHERE note_id=?",
                     (record.evidence_id,),
                 ).fetchone()
+            if grant is not None and grant[0] is not None:
+                raise ValueError("memory_add_source_revoked")
             if existing is not None:
                 current = self.require(record.evidence_id)
                 if current.verbatim != record.verbatim:

memhub/apps/core/src/one_memory/memory_server/durable_queue.py

@@ -24,10 +24,11 @@ from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 from uuid import uuid4
 
 from one_memory.background_work import in_the_background
+from one_memory.conversation_processing import TURN_IDLE_FLUSH_SECONDS, turn_batches
 from one_memory.database_access import main_database
 from one_memory.diagnostic_log import diagnostic_event
 from one_memory.imports.models import EvidenceOrigin, EvidenceRecord
@@ -894,16 +895,30 @@ class DurableWriteQueue:
         priority: str,
         connection: sqlite3.Connection,
     ) -> None:
-        """Append fresh-install input to an assembling queue row.
+        """Append input to its conversation's assembling queue row.
 
-        The row is not claimable until the scanner commits EOF and finalizes
-        it. No Evidence table is involved.
+        The row is not claimable. `finalize_sdk_file` turns what is complete
+        into write jobs (`turn_batches`) and keeps the rest here for the next
+        increment; `flush_idle_assemblies` writes a live row's rest once the
+        conversation has been quiet. `updated_at` moves only when a record is
+        new or revised, so a re-read does not keep a quiet conversation open.
+        No Evidence table is involved.
         """
         assembly_key = (
             f"assembly:{record.user_id}:{record.source_id}:"
             f"{hashlib.sha256(file_key.encode()).hexdigest()}:"
             f"{hashlib.sha256(record.conversation_id.encode()).hexdigest()}"
         )
+        session_key = internal_session_id_for(record.conversation_id, record.source_id)
+        written = connection.execute(
+            "SELECT 1 FROM memory_write_jobs AS job, json_each(job.evidence_ids_json) AS ref "
+            "WHERE job.session_key=? AND job.state!='assembling' AND ref.value=? LIMIT 1",
+            (session_key, record.evidence_id),
+        ).fetchone()
+        if written is not None:
+            # A re-read of a record a job already holds: staging it again
+            # would regroup it with later turns into a second job.
+            return
         row = connection.execute(
             "SELECT job_id,payload_json,priority FROM memory_write_jobs "
             "WHERE idempotency_key=? AND state='assembling'",
@@ -914,8 +929,20 @@ class DurableWriteQueue:
             payload = json.loads(str(row["payload_json"]))
             records = list(payload.get("input_records") or [])
         dumped = record.model_dump(mode="json")
-        if not any(item.get("evidence_id") == record.evidence_id for item in records):
+        index = next(
+            (
+                position
+                for position, item in enumerate(records)
+                if item.get("evidence_id") == record.evidence_id
+            ),
+            None,
+        )
+        changed = index is None or records[index] != dumped
+        if index is None:
             records.append(dumped)
+        elif changed:
+            # A revision replaces the carried copy in place.
+            records[index] = dumped
         now = datetime.now(UTC).isoformat()
         payload_json = json.dumps(
             {
@@ -926,15 +953,13 @@ class DurableWriteQueue:
             sort_keys=True,
         )
         if row is not None:
+            merged_priority = "live" if "live" in (row["priority"], priority) else "normal"
+            if not changed and merged_priority == row["priority"]:
+                return
             connection.execute(
-                "UPDATE memory_write_jobs SET payload_json=?,priority=?,updated_at=? "
-                "WHERE job_id=?",
-                (
-                    payload_json,
-                    "live" if row["priority"] == "live" or priority == "live" else "normal",
-                    now,
-                    str(row["job_id"]),
-                ),
+                "UPDATE memory_write_jobs SET payload_json=?,priority=?,"
+                "updated_at=CASE WHEN ? THEN ? ELSE updated_at END WHERE job_id=?",
+                (payload_json, merged_priority, int(changed), now, str(row["job_id"])),
             )
             return
         queue_seq = int(
@@ -952,7 +977,7 @@ class DurableWriteQueue:
             (
                 str(uuid4()),
                 queue_seq,
-                internal_session_id_for(record.conversation_id, record.source_id),
+                session_key,
                 record.source_id,
                 assembly_key,
                 assembly_key,
@@ -982,92 +1007,215 @@ class DurableWriteQueue:
         source_id: str,
         file_key: str,
         connection: sqlite3.Connection,
+        flush: bool = True,
     ) -> tuple[str, ...]:
+        """Write what one file's assembling rows hold that the turn rule completes.
+
+        `flush` (history at end of file) completes the last turn and writes a
+        short batch; without it (live polling) those stay in the row.
+        """
         rows = connection.execute(
             "SELECT * FROM memory_write_jobs WHERE state='assembling' "
             "AND source_key=? ORDER BY queue_seq",
             (source_id,),
         ).fetchall()
-        finalized = []
+        finalized: list[str] = []
         for row in rows:
             payload = json.loads(str(row["payload_json"]))
             if payload.get("file_key") != file_key:
                 continue
-            records = tuple(
-                EvidenceRecord.model_validate(item)
-                for item in payload.get("input_records") or []
-                if item.get("user_id") == user_id
+            finalized.extend(
+                self._write_assembled_turns(connection, row, user_id=user_id, flush=flush)
             )
-            if not records:
-                connection.execute(
-                    "DELETE FROM memory_write_jobs WHERE job_id=?",
-                    (str(row["job_id"]),),
+        return tuple(finalized)
+
+    def assemble_live_records(
+        self,
+        records: tuple[EvidenceRecord, ...],
+        *,
+        file_key: str,
+    ) -> tuple[str, ...]:
+        """Stage live records and write what is complete, in one transaction.
+
+        For live capture outside the import runner (host hooks): the same
+        assembling rows, turn rule and idle flush as live file polling.
+        """
+        if not records:
+            return ()
+        first = records[0]
+        with (
+            main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
+            connection,
+        ):
+            for record in records:
+                self.stage_sdk_input(
+                    record, file_key=file_key, priority="live", connection=connection
                 )
-                continue
-            first = records[0]
-            session_id = internal_session_id_for(
-                first.conversation_id,
-                first.source_id,
-            )
-            material = ":".join(f"{item.evidence_id}:{item.content_hash}" for item in records)
-            request_id = f"turn:v2:{hashlib.sha256(material.encode()).hexdigest()}"
-            existing = connection.execute(
-                "SELECT job_id FROM memory_write_jobs WHERE idempotency_key=? AND job_id!=?",
-                (request_id, str(row["job_id"])),
-            ).fetchone()
-            if existing is not None:
-                connection.execute(
-                    "DELETE FROM memory_write_jobs WHERE job_id=?",
-                    (str(row["job_id"]),),
-                )
-                finalized.append(str(existing["job_id"]))
-                continue
-            request = WriteRequest(
-                internal_session_id=session_id,
+            return self.finalize_sdk_file(
+                user_id=first.user_id,
                 source_id=first.source_id,
-                batch_identity=BatchIdentity(
-                    internal_session_id=session_id,
-                    first_round=1,
-                    last_round=1,
-                ),
-                evidence_ids=[item.evidence_id for item in records],
-                input_records=records,
-                idempotency_key=request_id,
-                priority="live" if str(row["priority"]) == "live" else "normal",
-                event_at=max(
-                    (item.created_at for item in records if item.created_at),
-                    default=first.imported_at,
-                ),
-                time_quality="exact" if any(item.created_at for item in records) else "approximate",
+                file_key=file_key,
+                connection=connection,
+                flush=False,
             )
+
+    def flush_idle_assemblies(self, *, now: datetime | None = None) -> tuple[str, ...]:
+        """Write the rest of every live conversation quiet for 10 minutes.
+
+        Its last turn counts as complete and a batch short of 5 turns is
+        written. History rows are left alone: their end of file flushes them.
+        """
+        current = now or datetime.now(UTC)
+        cutoff = (current - timedelta(seconds=TURN_IDLE_FLUSH_SECONDS)).isoformat()
+        query = (
+            "SELECT * FROM memory_write_jobs WHERE state='assembling' AND priority='live' "
+            "AND updated_at<=? ORDER BY queue_seq"
+        )
+        # A read first: most ticks find nothing and must not take the writer.
+        with main_database(self.database_path).read(row_factory=sqlite3.Row) as connection:
+            if connection.execute(query, (cutoff,)).fetchone() is None:
+                return ()
+        finalized: list[str] = []
+        with (
+            main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
+            connection,
+        ):
+            for row in connection.execute(query, (cutoff,)).fetchall():
+                finalized.extend(self._write_assembled_turns(connection, row, flush=True))
+        return tuple(finalized)
+
+    def _write_assembled_turns(
+        self,
+        connection: sqlite3.Connection,
+        row: sqlite3.Row,
+        *,
+        flush: bool,
+        user_id: str | None = None,
+    ) -> list[str]:
+        """Turn one assembling row's complete batches into jobs; keep the carry."""
+        payload = json.loads(str(row["payload_json"]))
+        records = tuple(
+            EvidenceRecord.model_validate(item)
+            for item in payload.get("input_records") or []
+            if user_id is None or item.get("user_id") == user_id
+        )
+        grouped = turn_batches(records, flush=flush)
+        priority: Literal["live", "normal"] = "live" if str(row["priority"]) == "live" else "normal"
+        job_ids = [
+            self._insert_turn_batch(connection, batch, priority=priority)
+            for batch in grouped.batches
+        ]
+        if not grouped.carry:
             connection.execute(
-                "UPDATE memory_write_jobs SET session_key=?,batch_id=?,"
-                "idempotency_key=?,payload_json=?,evidence_ids_json=?,state='pending',"
-                "event_at=?,time_quality=?,updated_at=? WHERE job_id=?",
+                "DELETE FROM memory_write_jobs WHERE job_id=?", (str(row["job_id"]),)
+            )
+        elif grouped.batches or len(grouped.carry) != len(records):
+            # `updated_at` stays: the idle clock counts new messages only.
+            connection.execute(
+                "UPDATE memory_write_jobs SET payload_json=? WHERE job_id=?",
                 (
-                    session_id,
-                    request.batch_identity.batch_id,
-                    request_id,
                     json.dumps(
-                        request.model_dump(
-                            mode="json",
-                            exclude={"event_at", "time_quality"},
-                        ),
+                        {
+                            **payload,
+                            "input_records": [
+                                record.model_dump(mode="json") for record in grouped.carry
+                            ],
+                        },
                         sort_keys=True,
                     ),
-                    json.dumps(list(request.evidence_ids)),
-                    (
-                        request.event_at.isoformat()
-                        if request.event_at
-                        else datetime.now(UTC).isoformat()
-                    ),
-                    request.time_quality,
-                    datetime.now(UTC).isoformat(),
                     str(row["job_id"]),
                 ),
             )
-            finalized.append(str(row["job_id"]))
-        return tuple(finalized)
+        return job_ids
+
+    def _insert_turn_batch(
+        self,
+        connection: sqlite3.Connection,
+        records: tuple[EvidenceRecord, ...],
+        *,
+        priority: Literal["live", "normal"],
+    ) -> str:
+        """One pending job for one batch of whole turns, keyed by its content.
+
+        Same records, same key: a re-read returns the job already written.
+        """
+        first = records[0]
+        session_id = internal_session_id_for(first.conversation_id, first.source_id)
+        material = ":".join(f"{item.evidence_id}:{item.content_hash}" for item in records)
+        request_id = f"turn:v2:{hashlib.sha256(material.encode()).hexdigest()}"
+        existing = connection.execute(
+            "SELECT job_id FROM memory_write_jobs WHERE idempotency_key=?",
+            (request_id,),
+        ).fetchone()
+        if existing is not None:
+            return str(existing["job_id"])
+        request = WriteRequest(
+            internal_session_id=session_id,
+            source_id=first.source_id,
+            batch_identity=BatchIdentity(
+                internal_session_id=session_id,
+                first_round=1,
+                last_round=1,
+            ),
+            evidence_ids=[item.evidence_id for item in records],
+            input_records=records,
+            idempotency_key=request_id,
+            priority=priority,
+        )
+        # The batch's time is its first user message's, as for every other job.
+        event_at, time_quality = self._event_time_for_request(request)
+        now = datetime.now(UTC)
+        job_id = str(uuid4())
+        queue_seq = int(
+            connection.execute(
+                "SELECT COALESCE(MAX(queue_seq),0)+1 FROM memory_write_jobs"
+            ).fetchone()[0]
+        )
+        event = event_at or now
+        if event.tzinfo is None:
+            event = event.replace(tzinfo=UTC)
+        connection.execute(
+            "INSERT INTO memory_write_jobs ("
+            "job_id,queue_seq,session_key,source_key,batch_id,idempotency_key,"
+            "payload_json,evidence_ids_json,state,priority,attempts,max_attempts,"
+            "lease_owner,lease_expires_at,hy_invoked,last_error,event_at,time_quality,"
+            "created_at,updated_at,sdk_raw_ids_json,sdk_memory_ids_json"
+            ") VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+            (
+                job_id,
+                queue_seq,
+                session_id,
+                first.source_id,
+                request.batch_identity.batch_id,
+                request_id,
+                json.dumps(
+                    request.model_dump(mode="json", exclude={"event_at", "time_quality"}),
+                    sort_keys=True,
+                ),
+                json.dumps(list(request.evidence_ids)),
+                PENDING,
+                priority,
+                0,
+                3,
+                None,
+                None,
+                0,
+                None,
+                event.astimezone(UTC).isoformat(),
+                time_quality,
+                now.isoformat(),
+                now.isoformat(),
+                "[]",
+                "[]",
+            ),
+        )
+        if not self.sdk_authority:
+            # Legacy catch-up reads coverage from these references.
+            connection.executemany(
+                "INSERT INTO memory_write_job_evidence (job_id, evidence_id) VALUES (?,?)",
+                ((job_id, evidence_id) for evidence_id in request.evidence_ids),
+            )
+        return job_id
 
     def _freeze_request_inputs(self, request: WriteRequest) -> WriteRequest:
         if request.input_records or self.evidence is None:
@@ -2500,6 +2648,9 @@ class DurableWriteDrain:
         #: a model outage clears on its own, and a queue that never goes idle
         #: must still re-admit the write once the wait has elapsed.
         self._last_model_readmit: datetime | None = None
+        #: A live conversation quiet for 10 minutes has its last turn and short
+        #: batch written (`flush_idle_assemblies`); checked on the same timer.
+        self._last_assembly_flush: datetime | None = None
         #: Idle polls do not need to hit the single writer every 50ms. A
         #: 1m Evidence table makes a miss expensive enough that the tail
         #: call-event write can be behind it; back off only while idle and
@@ -3032,6 +3183,15 @@ class DurableWriteDrain:
                         self.queue.readmit_model_waiters,
                         operation_name="readmit_model_waiters",
                     )
+                if (
+                    self._last_assembly_flush is None
+                    or tick_started - self._last_assembly_flush >= self._reap_interval
+                ):
+                    self._last_assembly_flush = tick_started
+                    await self._run_idle_maintenance(
+                        self.queue.flush_idle_assemblies,
+                        operation_name="flush_idle_assemblies",
+                    )
                 if self.reconcile_unknown is not None and (
                     self._last_unknown_reconcile is None
                     or tick_started - self._last_unknown_reconcile

memhub/apps/core/src/one_memory/model_profiles.py

@@ -1136,20 +1136,31 @@ class ModelProfileStore:
         )
 
 
+def disable_thinking_params(
+    profile: ModelProfile | ModelProfileSnapshot,
+) -> dict[str, object]:
+    """Provider-specific thinking-off params, empty unless the profile opts in.
+
+    The one mapping for every request Core builds itself and for the SDK's
+    `llm.extra_body`, so a model that thinks on one path thinks on none.
+    """
+    if not profile.disable_thinking:
+        return {}
+    if profile.provider_kind is ProviderKind.OPENAI_COMPATIBLE:
+        if urlparse(profile.base_url or "").hostname == "api.deepseek.com":
+            return {"thinking": {"type": "disabled"}}
+        return {"reasoning_effort": "none"}
+    if profile.provider_kind is ProviderKind.OLLAMA:
+        return {"think": False}
+    return {}
+
+
 def attach_disable_thinking(
     profile: ModelProfile | ModelProfileSnapshot,
     payload: dict[str, object],
 ) -> dict[str, object]:
     """Attach provider-specific thinking-off params when the profile opts in."""
-    if not profile.disable_thinking:
-        return payload
-    if profile.provider_kind is ProviderKind.OPENAI_COMPATIBLE:
-        if urlparse(profile.base_url or "").hostname == "api.deepseek.com":
-            payload["thinking"] = {"type": "disabled"}
-        else:
-            payload["reasoning_effort"] = "none"
-    elif profile.provider_kind is ProviderKind.OLLAMA:
-        payload["think"] = False
+    payload.update(disable_thinking_params(profile))
     return payload
 
 

memhub/apps/core/src/one_memory/sdk_migration_log.py

@@ -125,6 +125,31 @@ def describe_error(error: BaseException) -> dict[str, object]:
     return {"chain": chain}
 
 
+_PY_FRAME = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.\-]*\.py:\d{1,7}:[A-Za-z_][A-Za-z0-9_]*$")
+
+
+def error_frames(error: BaseException, *, limit: int = 16) -> tuple[str, ...]:
+    """The raised error's innermost `file.py:line:function` frames, for a stack field.
+
+    Library frames are kept (a KeyError raised inside Chroma is located by
+    them); `<module>`-style names are dropped because stack fields refuse them.
+    """
+    chain = describe_error(error)["chain"]
+    frames = chain[0].get("frames", ()) if chain else ()  # type: ignore[index,union-attr]
+    return tuple(frame for frame in frames if _PY_FRAME.fullmatch(frame))[-limit:]
+
+
+def error_chain_text(error: BaseException) -> str:
+    """`describe_error` as one log token: `Type@file:line:function>...;Cause@...`.
+
+    For a plain text log (core.log). Types and frames only, never a message.
+    """
+    return ";".join(
+        f"{entry['type']}@{'>'.join(entry.get('frames', ()))}"  # type: ignore[arg-type]
+        for entry in describe_error(error)["chain"]  # type: ignore[union-attr]
+    )
+
+
 def _build() -> dict[str, object]:
     from one_memory import __version__
 

memhub/apps/core/src/one_memory/sdk_migration_runner.py

@@ -829,6 +829,7 @@ def run_startup_migration(
                 ignore_errors=True,
             )
         discard_failed_state(data_dir)
+        _repair_metadata_after_cutover(data_dir, database_path, log)
         progress("cleanup")
         with log.step("cleanup.drop_legacy_tables") as step:
             dropped = resume_legacy_cleanup(data_dir=data_dir, database_path=database_path)
@@ -847,6 +848,25 @@ def run_startup_migration(
     )
 
 
+def _repair_metadata_after_cutover(data_dir: Path, database_path: Path, log: _StepLog) -> None:
+    """Restore lost metadata here, where Runtime is stopped and no Chroma client is open.
+
+    Writes the repair receipt, so Core's startup fallback finds it and does
+    nothing. Authority is already committed: a failure is logged and left to
+    that fallback instead of failing the migration.
+    """
+    from one_memory.hy_sdk_migration import repair_sdk_metadata_from_bundle
+
+    try:
+        with log.step("cutover.metadata_repair") as step:
+            repair = repair_sdk_metadata_from_bundle(data_dir=data_dir, database_path=database_path)
+            step["mapped"] = repair.mapped
+            step["vector_rows"] = repair.vector_rows
+            step["coding_rows"] = repair.coding_rows
+    except Exception:
+        return
+
+
 def _restore_after_failure(
     *,
     data_dir: Path,

memhub/apps/core/src/one_memory/sdk_write_processor.py

@@ -16,6 +16,7 @@ from one_memory.secrets import SecretStore, SecretStoreUnavailableError
 
 PRODUCT_AGENT_ID = "all-apps"
 _MESSAGE_ROLES = frozenset({"user", "assistant", "tool"})
+_CONVERSATION_ROLES = frozenset({"user", "assistant"})
 
 
 @dataclass(frozen=True)
@@ -41,12 +42,23 @@ class SdkWriteProcessor:
     ) -> SdkWriteReceipt:
         if not records:
             raise RuntimeError("hy_memory_task_without_inputs")
+        extracted = all(_is_extracted_content(record) for record in records)
+        if not extracted and not any(
+            record.content_kind is EvidenceContentKind.RAW_CONVERSATION
+            and record.role in _CONVERSATION_ROLES
+            for record in records
+        ):
+            # Nothing to remember: tool records alone. The SDK strips them on
+            # the chat path and answers 400 "content or messages is
+            # required", which used to settle the job as unknown. Completed
+            # with no memories, and the SDK is never called.
+            return SdkWriteReceipt()
 
         scope = _write_scope(records)
         metadata = _write_metadata(records, scope)
         client = await self._client_handle.client()
 
-        if all(_is_extracted_content(record) for record in records):
+        if extracted:
             contents = [sanitize_model_input(record.verbatim) for record in records]
             await before_hy_invocation()
             result = await asyncio.to_thread(

memhub/apps/core/tests/test_conversation_turn_grouping.py

@@ -1,90 +1,199 @@
-"""A proposal and the turn that adopts it must reach HY as one turn."""
+"""The turn rule: what a conversation's records become before they reach HY.
+
+A turn is one user message and everything up to the next one. Every 5
+complete turns are written together, sooner when the next turn would pass
+8000 characters; a short batch waits for the end of a history file or 10
+quiet minutes. No batch is written without a user message.
+"""
 
 from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from one_memory.conversation_processing import _conversation_chunks
+from one_memory.conversation_processing import (
+    TURN_BATCH_CHARACTERS,
+    TURN_BATCH_TURNS,
+    conversation_turns,
+    turn_batches,
+)
 from one_memory.imports.models import EvidenceInput, EvidenceOrigin, OriginMetadata, WrittenBy
 from one_memory.imports.storage import EvidenceStore
 
+_WRITER = {"user": "person", "assistant": "agent", "tool": "tool"}
 
-def _records(tmp_path, *speakers: tuple[str, str, str]) -> list:
+
+def _records(tmp_path, *speakers: tuple[str, str]) -> list:
     store = EvidenceStore(tmp_path / "one-memory.sqlite3")
-    records = []
-    for index, (role, written_by, text) in enumerate(speakers):
-        records.append(
-            store.put(
-                EvidenceInput(
-                    source_id="cursor",
-                    conversation_id="conversation",
-                    message_id=f"m{index}",
-                    role=role,
-                    written_by=WrittenBy(written_by),
-                    verbatim=text,
-                    created_at=datetime.now(UTC),
-                    origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
-                )
-            ).record
-        )
-    return records
-
-
-def test_an_unanswered_assistant_prefix_joins_the_turn_that_adopts_it(tmp_path) -> None:
-    """`assistant` proposal -> `user` explicit adoption is one turn, not two.
-
-    Splitting it sends HY an answer without what it answers, and loses the
-    authority pair the author/adoption quality gate depends on.
-    """
-    records = _records(
-        tmp_path,
-        ("assistant", "agent", "建议把超时改成 30 秒。"),
-        ("user", "person", "就这么做，超时改成 30 秒。"),
-    )
-
-    chunks = _conversation_chunks(records)
-
-    assert [[record.role for record in chunk] for chunk in chunks] == [["assistant", "user"]]
-
-
-def test_a_lone_user_message_does_not_become_a_model_input(tmp_path) -> None:
-    """A user message with no reply yet is Evidence, not a completed turn.
-
-    Concurrent-import fixtures write `user` then `assistant` pairs, but a lone
-    user message must not create an HY task: that is the complete-turn rule,
-    not a queue stall.
-    """
-    records = _records(tmp_path, ("user", "person", "把超时改成 30 秒。"))
-
-    assert _conversation_chunks(records, completed_only=True) == []
-
-
-def test_only_the_unanswered_trailing_turn_is_held_back(tmp_path) -> None:
-    """Earlier complete turns stay eligible while the open tail waits."""
-    records = _records(
-        tmp_path,
-        ("user", "person", "先把超时改成 30 秒。"),
-        ("assistant", "agent", "已经改好了。"),
-        ("user", "person", "接着把发布包传上去。"),
-    )
-
-    assert _conversation_chunks(records, completed_only=True) == [(records[0], records[1])]
-
-
-def test_the_complete_turn_contract_used_for_concurrent_import_reds(tmp_path) -> None:
-    """Node-id level answer for `[BLOCK C2 -> C1]` concurrent-import reds.
-
-    Those fixtures write `user` turns and never emit the closing
-    `assistant`/tool event, so under the complete-turn rule no HY task is
-    created inside the 5s window. That is the contract, not a queue stall:
-    Evidence is still committed and searchable, and the task appears once the
-    closing event arrives.
-    """
-    closed = _records(
-        tmp_path,
-        ("user", "person", "把超时改成 30 秒。"),
-        ("assistant", "agent", "已改为 30 秒。"),
-    )
-    assert [[record.role for record in chunk] for chunk in _conversation_chunks(closed)] == [
-        ["user", "assistant"]
+    return [
+        store.put(
+            EvidenceInput(
+                source_id="cursor",
+                conversation_id="conversation",
+                message_id=f"m{index}",
+                role=role,
+                written_by=WrittenBy(_WRITER[role]),
+                verbatim=text,
+                created_at=datetime.now(UTC),
+                origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+            )
+        ).record
+        for index, (role, text) in enumerate(speakers)
     ]
+
+
+def _turns(count: int, *, answer: str = "好的。") -> list[tuple[str, str]]:
+    return [
+        part
+        for index in range(count)
+        for part in (("user", f"第 {index} 个问题"), ("assistant", answer))
+    ]
+
+
+def _roles(batches) -> list[list[str]]:
+    return [[record.role for record in batch] for batch in batches]
+
+
+def test_rule_constants_are_the_product_rule() -> None:
+    assert (TURN_BATCH_TURNS, TURN_BATCH_CHARACTERS) == (5, 8_000)
+
+
+def test_a_turn_runs_from_one_user_message_to_the_next(tmp_path) -> None:
+    records = _records(
+        tmp_path,
+        ("user", "跑一下测试"),
+        ("assistant", "好"),
+        ("tool", "3 passed"),
+        ("assistant", "都通过了"),
+        ("user", "提交吧"),
+        ("assistant", "已提交"),
+    )
+
+    assert [len(turn) for turn in conversation_turns(records)] == [4, 2]
+
+
+def test_records_before_the_first_user_message_join_the_first_turn(tmp_path) -> None:
+    """An assistant proposal and the user turn adopting it reach HY together."""
+    records = _records(
+        tmp_path,
+        ("assistant", "建议把超时改成 30 秒。"),
+        ("user", "就这么做，超时改成 30 秒。"),
+        ("assistant", "改好了。"),
+        ("user", "再把重试改成 3 次。"),
+        ("assistant", "也改好了。"),
+    )
+
+    assert _roles(conversation_turns(records)) == [
+        ["assistant", "user", "assistant"],
+        ["user", "assistant"],
+    ]
+
+
+def test_five_complete_turns_are_one_batch_and_the_open_turn_waits(tmp_path) -> None:
+    records = _records(tmp_path, *_turns(6))
+
+    grouped = turn_batches(records, flush=False)
+
+    assert [len(batch) for batch in grouped.batches] == [10]
+    assert grouped.batches[0] == tuple(records[:10])
+    assert grouped.carry == tuple(records[10:])
+
+
+def test_complete_turns_short_of_five_wait_for_flush(tmp_path) -> None:
+    records = _records(tmp_path, *_turns(3))
+
+    waiting = turn_batches(records, flush=False)
+    assert waiting.batches == ()
+    assert waiting.carry == tuple(records)
+
+    flushed = turn_batches(records, flush=True)
+    assert flushed.batches == (tuple(records),)
+    assert flushed.carry == ()
+
+
+def test_a_lone_user_message_is_not_written_until_flush(tmp_path) -> None:
+    records = _records(tmp_path, ("user", "把超时改成 30 秒。"))
+
+    assert turn_batches(records, flush=False).batches == ()
+    assert turn_batches(records, flush=True).batches == (tuple(records),)
+
+
+def test_the_turn_that_would_pass_8000_characters_starts_the_next_batch(tmp_path) -> None:
+    answer = "句" * 3_000
+    records = _records(tmp_path, *_turns(4, answer=answer))
+
+    grouped = turn_batches(records, flush=False)
+
+    # Turns 1+2 fit; turn 3 would pass 8000, so 1+2 are written now. Turn 3
+    # is complete but alone, and turn 4 is still open.
+    assert grouped.batches == (tuple(records[:4]),)
+    assert grouped.carry == tuple(records[4:])
+    assert sum(len(record.verbatim) for record in grouped.batches[0]) <= TURN_BATCH_CHARACTERS
+
+
+def test_one_turn_over_8000_characters_is_a_batch_of_its_own_and_not_split(tmp_path) -> None:
+    records = _records(
+        tmp_path,
+        ("user", "先问个小问题"),
+        ("assistant", "好"),
+        ("user", "写一份方案"),
+        ("assistant", "段落。" * 7_000),
+        ("user", "谢谢"),
+    )
+
+    grouped = turn_batches(records, flush=False)
+
+    assert grouped.batches == (tuple(records[:2]), tuple(records[2:4]))
+    assert grouped.carry == (records[4],)
+    assert all(":segment:" not in record.evidence_id for record in grouped.batches[1])
+
+
+def test_tool_output_counts_only_what_hy_reads(tmp_path) -> None:
+    """A tool record counts up to `_HY_TOOL_CHARACTERS`, not its full text."""
+    from one_memory.conversation_processing import _HY_TOOL_CHARACTERS
+
+    parts = [
+        part
+        for index in range(5)
+        for part in (("user", f"第 {index} 步"), ("tool", "x" * 50_000), ("assistant", "完成"))
+    ]
+    records = _records(tmp_path, *parts, ("user", "下一步"))
+
+    grouped = turn_batches(records, flush=False)
+
+    assert _HY_TOOL_CHARACTERS == 512
+    assert [len(batch) for batch in grouped.batches] == [15]
+
+
+def test_records_with_no_user_message_are_never_a_batch(tmp_path) -> None:
+    """The SDK strips tool records on the chat path and refuses what is left.
+
+    A tool-only batch came back 400 "content or messages is required" and
+    was recorded unknown, 16 times. Carried until a user message arrives;
+    dropped when the conversation is flushed without one.
+    """
+    records = _records(tmp_path, ("tool", "ls output"), ("assistant", "看完了"))
+
+    waiting = turn_batches(records, flush=False)
+    assert waiting.batches == ()
+    assert waiting.carry == tuple(records)
+    assert turn_batches(records, flush=True) == turn_batches((), flush=True)
+
+    later = _records(tmp_path, ("user", "那就开始吧"), ("assistant", "开始"))
+    joined = turn_batches((*waiting.carry, *later), flush=True)
+    assert _roles(joined.batches) == [["tool", "assistant", "user", "assistant"]]
+
+
+def test_every_batch_holds_a_user_message(tmp_path) -> None:
+    records = _records(
+        tmp_path,
+        ("tool", "前置输出"),
+        *_turns(7, answer="答" * 2_500),
+        ("tool", "尾部输出"),
+    )
+
+    for flush in (False, True):
+        grouped = turn_batches(records, flush=flush)
+        assert grouped.batches
+        assert all(any(record.role == "user" for record in batch) for batch in grouped.batches)
+        written = [record for batch in grouped.batches for record in batch]
+        assert written + list(grouped.carry) == records

memhub/apps/core/tests/test_core_ready_before_recovery.py

@@ -142,3 +142,112 @@ def test_shutdown_during_recovery_stops_it_without_starting_more(tmp_path, monke
             release.set()
 
     asyncio.run(scenario())
+
+
+def test_metadata_repair_never_overlaps_an_sdk_client(tmp_path, monkeypatch):
+    """The repair's own Chroma client dropped the SDK's shared System (KeyError).
+
+    The first library request opened the SDK client while the repair ran
+    after readiness. Now every SDK client of the home waits for the repair.
+    """
+    from one_memory import hy_sdk_runtime
+    from one_memory.hy_sdk_runtime import ClientLease
+
+    release = threading.Event()
+    entered = threading.Event()
+    events: list[str] = []
+
+    def slow_repair(**_):
+        entered.set()
+        events.append("repair_started")
+        release.wait(10)
+        events.append("repair_finished")
+        return MetadataRepairResult(None, 0, 0, 0)
+
+    def open_client(profile, *, api_key, data_dir, http_transport_factory):
+        events.append("client_opened")
+        return ClientLease(object(), data_dir.resolve())
+
+    monkeypatch.setattr(hy_sdk_migration, "repair_sdk_metadata_from_bundle", slow_repair)
+    monkeypatch.setattr(hy_sdk_runtime, "_acquire_client", open_client)
+    monkeypatch.setattr(ClientLease, "close", lambda self: None)
+
+    async def scenario():
+        app = application(tmp_path)
+        try:
+            async with app.router.lifespan_context(app), client_for(app) as client:
+                assert await asyncio.to_thread(entered.wait, 5)
+                handle = hy_sdk_runtime.SdkClientHandle(
+                    profile=hy_sdk_runtime.builtin_retrieval_snapshot(),
+                    api_key=None,
+                    data_dir=app.state.settings.data_dir,
+                )
+                opening = asyncio.create_task(handle.client())
+                await asyncio.sleep(0.2)
+                # Readiness and reads answer; the SDK client waits.
+                assert (await health(client))["status"] == "ok"
+                assert not opening.done()
+                release.set()
+                await asyncio.wait_for(opening, 5)
+                await wait_for_recovery(client)
+        finally:
+            release.set()
+
+    asyncio.run(scenario())
+    assert events == ["repair_started", "repair_finished", "client_opened"]
+
+
+def test_stores_are_not_held_beside_an_open_sdk_client(tmp_path):
+    from one_memory import hy_sdk_runtime
+
+    key = tmp_path.resolve()
+    with hy_sdk_runtime._STORE_GATE:
+        hy_sdk_runtime._OPENING[key] = 1
+    try:
+        assert hy_sdk_runtime.hold_stores_offline(tmp_path) is False
+    finally:
+        with hy_sdk_runtime._STORE_GATE:
+            del hy_sdk_runtime._OPENING[key]
+    assert hy_sdk_runtime.hold_stores_offline(tmp_path) is True
+    assert hy_sdk_runtime.hold_stores_offline(tmp_path) is False
+    hy_sdk_runtime.release_stores(tmp_path)
+    hy_sdk_runtime.release_stores(tmp_path)
+    assert hy_sdk_runtime.hold_stores_offline(tmp_path) is True
+    hy_sdk_runtime.release_stores(tmp_path)
+
+
+def test_a_failed_recovery_step_records_where_it_failed(tmp_path, monkeypatch, caplog):
+    """`error_type=KeyError` alone named no location; frames, never the message."""
+    from one_memory import api
+
+    def lookup(systems):
+        return systems["private-identifier"]
+
+    def broken_repair(**_):
+        return lookup({})
+
+    events: list[dict] = []
+    monkeypatch.setattr(hy_sdk_migration, "repair_sdk_metadata_from_bundle", broken_repair)
+    monkeypatch.setattr(api, "diagnostic_event", lambda **values: events.append(values))
+
+    async def scenario():
+        app = application(tmp_path)
+        async with app.router.lifespan_context(app), client_for(app) as client:
+            await wait_for_recovery(client)
+
+    with caplog.at_level("WARNING"):
+        asyncio.run(scenario())
+
+    [line] = [
+        record.getMessage()
+        for record in caplog.records
+        if "core_recovery_step_failed" in record.getMessage()
+    ]
+    assert "step=sdk_metadata_repair error_type=KeyError chain=KeyError@" in line
+    assert "test_core_ready_before_recovery.py:" in line and ":lookup" in line
+    assert "private-identifier" not in line
+    [failed] = [event for event in events if event.get("event") == "core.recovery_step_failed"]
+    assert failed["error_type"] == "KeyError"
+    assert failed["safe_stack"][-1].startswith("test_core_ready_before_recovery.py:")
+    assert failed["safe_stack"][-1].endswith(":lookup")
+    assert "private-identifier" not in repr(failed)

memhub/apps/core/tests/test_disable_thinking.py

@@ -8,6 +8,7 @@ import json
 import httpx
 from pydantic import SecretStr
 
+from one_memory.hy_sdk_runtime import build_sdk_config
 from one_memory.imports.models import (
     EvidenceInput,
     EvidenceOrigin,
@@ -17,6 +18,7 @@ from one_memory.imports.models import (
 from one_memory.imports.storage import EvidenceStore
 from one_memory.memory_batch_extraction import MemoryBatchExtractor
 from one_memory.memory_qa import _answer_request
+from one_memory.model_probe import _structured_request
 from one_memory.model_profiles import (
     ModelProfileInput,
     ModelProfileStore,
@@ -243,3 +245,94 @@ def test_ollama_batch_sends_think_false_only_when_disabled(tmp_path) -> None:
     )
     assert payloads[0]["think"] is False
     assert "reasoning_effort" not in payloads[0]
+
+
+def _deepseek(store, secrets, *, disable_thinking: bool):
+    return store.create(
+        ModelProfileInput(
+            name=f"deepseek-{disable_thinking}",
+            provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+            base_url="https://api.deepseek.com/v1",
+            model="deepseek-v4-flash",
+            api_key=SecretStr("sk-test"),
+            embedding_model="embedding-model",
+            embedding_provider_kind=ProviderKind.LOCAL_ONNX,
+            disable_thinking=disable_thinking,
+        ),
+        secrets,
+    )
+
+
+def test_sdk_config_turns_deepseek_thinking_off_like_the_probe(tmp_path) -> None:
+    """DeepSeek Flash spent the SDK's 2000 agent tokens thinking: EMPTY_RESPONSE.
+
+    The SDK sends `llm.extra_body` on every chat call (chat and the shared
+    coding/working provider), and the probe must ask the model the same way.
+    """
+    database = tmp_path / "one-memory.sqlite3"
+    secrets = InMemorySecretStore()
+    store = ModelProfileStore(database)
+    profile = _deepseek(store, secrets, disable_thinking=True)
+    snapshot = store.snapshot(profile.profile_id, profile.revision)
+
+    config = build_sdk_config(snapshot, api_key="sk-test", data_dir=tmp_path / "data")
+
+    assert config.llm.extra_body == {"thinking": {"type": "disabled"}}
+    _m, _u, _h, probe = _structured_request(profile, "sk-test")
+    assert probe["thinking"] == config.llm.extra_body["thinking"]
+    assert "reasoning_effort" not in probe
+
+
+def test_sdk_config_sends_no_thinking_fields_unless_the_profile_opts_in(tmp_path) -> None:
+    database = tmp_path / "one-memory.sqlite3"
+    secrets = InMemorySecretStore()
+    store = ModelProfileStore(database)
+    profile = _deepseek(store, secrets, disable_thinking=False)
+    snapshot = store.snapshot(profile.profile_id, profile.revision)
+
+    config = build_sdk_config(snapshot, api_key="sk-test", data_dir=tmp_path / "data")
+
+    assert not config.llm.extra_body
+
+
+def test_sdk_config_maps_other_providers_through_the_same_table(tmp_path) -> None:
+    database = tmp_path / "one-memory.sqlite3"
+    secrets = InMemorySecretStore()
+    store = ModelProfileStore(database)
+    openai = store.create(
+        ModelProfileInput(
+            name="openai",
+            provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+            base_url="https://api.example.com/v1",
+            model="gpt-test",
+            api_key=SecretStr("sk-test"),
+            embedding_model="embedding-model",
+            embedding_provider_kind=ProviderKind.LOCAL_ONNX,
+        ),
+        secrets,
+    )
+    ollama = store.create(
+        ModelProfileInput(
+            name="ollama",
+            provider_kind=ProviderKind.OLLAMA,
+            base_url="http://127.0.0.1:11434",
+            model="qwen",
+            embedding_model="embedding-model",
+            embedding_provider_kind=ProviderKind.LOCAL_ONNX,
+        ),
+        secrets,
+    )
+
+    openai_config = build_sdk_config(
+        store.snapshot(openai.profile_id, openai.revision),
+        api_key="sk-test",
+        data_dir=tmp_path / "data",
+    )
+    ollama_config = build_sdk_config(
+        store.snapshot(ollama.profile_id, ollama.revision),
+        api_key=None,
+        data_dir=tmp_path / "data",
+    )
+
+    assert openai_config.llm.extra_body == {"reasoning_effort": "none"}
+    assert ollama_config.llm.extra_body == {"think": False}

memhub/apps/core/tests/test_import_runner.py

@@ -773,3 +773,131 @@ def test_cancel_during_callback_keeps_a_replayable_pending_marker(tmp_path) -> N
 
     assert evidence_store.count() == 1
     assert len(_pending_markers(evidence_store.database_path)) == 1
+
+
+def said(message_id: str, role: str, text: str) -> EvidenceInput:
+    writer = {"user": WrittenBy.PERSON, "assistant": WrittenBy.AGENT, "tool": WrittenBy.TOOL}
+    return evidence(message_id).model_copy(
+        update={"role": role, "written_by": writer[role], "verbatim": text}
+    )
+
+
+def _sdk_runner(tmp_path):
+    database = tmp_path / "one-memory.sqlite3"
+    bootstrap_fresh_management_schema(database)
+    job_store = ImportJobStore(database)
+    queue = DurableWriteQueue(database, sdk_authority=True)
+    runner = ImportRunner(
+        evidence_store=RetiredEvidenceStore(database),  # type: ignore[arg-type]
+        job_store=job_store,
+        sdk_queue=queue,
+        sdk_authority=lambda: True,
+    )
+    return database, job_store, queue, runner
+
+
+def _read(runner, job_store, parts, *, operation: str, first_line: int = 1):
+    events: list[ScanEvent] = [
+        ScannedEvidence(evidence=part, cursor=cursor(f"line:{first_line + index}"))
+        for index, part in enumerate(parts)
+    ]
+    last = f"line:{first_line + len(parts) - 1}"
+    events.append(ScanCursorTouch(cursor=cursor(last).model_copy(update={"read_to_end": True})))
+    job = job_store.create(user_id="local", source_id="fixture", operation=operation)
+    asyncio.run(runner.run(adapter=FixtureAdapter(events), job_id=job.job_id))
+
+
+def _jobs(database) -> list[tuple[str, list[str]]]:
+    import json
+
+    with sqlite3.connect(database) as connection:
+        rows = connection.execute(
+            "SELECT state,payload_json FROM memory_write_jobs ORDER BY queue_seq"
+        ).fetchall()
+    return [
+        (state, [item["role"] for item in json.loads(payload)["input_records"]])
+        for state, payload in rows
+    ]
+
+
+def _turns(count: int, *, start: int = 0) -> list[EvidenceInput]:
+    return [
+        said(f"{role[0]}{index}", role, f"{role} {index}")
+        for index in range(start, start + count)
+        for role in ("user", "assistant")
+    ]
+
+
+def test_history_import_writes_five_turn_batches_and_flushes_the_rest_at_eof(tmp_path):
+    """A first import used to be one giant job per conversation."""
+    database, job_store, _queue, runner = _sdk_runner(tmp_path)
+
+    _read(runner, job_store, _turns(7), operation="history")
+
+    assert _jobs(database) == [
+        ("pending", ["user", "assistant"] * 5),
+        ("pending", ["user", "assistant"] * 2),
+    ]
+
+
+def test_live_polling_carries_an_open_turn_and_a_tool_only_increment(tmp_path):
+    """A transcript read mid-turn used to give tool-only jobs (16 unknown)."""
+    database, job_store, queue, runner = _sdk_runner(tmp_path)
+
+    _read(runner, job_store, _turns(2), operation="sync")
+    assert _jobs(database) == [("assembling", ["user", "assistant"] * 2)]
+
+    tools = [said("t1", "tool", "ls"), said("t2", "tool", "3 passed")]
+    _read(runner, job_store, tools, operation="sync", first_line=5)
+    assert _jobs(database) == [
+        ("assembling", ["user", "assistant", "user", "assistant", "tool", "tool"])
+    ]
+
+    # Ten quiet minutes: the open turn is complete and the short batch goes.
+    assert queue.flush_idle_assemblies(now=datetime.now(UTC) + timedelta(minutes=9)) == ()
+    flushed = queue.flush_idle_assemblies(now=datetime.now(UTC) + timedelta(minutes=11))
+    assert len(flushed) == 1
+    assert _jobs(database) == [
+        ("pending", ["user", "assistant", "user", "assistant", "tool", "tool"])
+    ]
+
+
+def test_live_polling_writes_five_complete_turns_and_keeps_the_open_one(tmp_path):
+    database, job_store, queue, runner = _sdk_runner(tmp_path)
+
+    _read(runner, job_store, _turns(5), operation="sync")
+    assert [state for state, _ in _jobs(database)] == ["assembling"]
+
+    _read(runner, job_store, [said("u5", "user", "user 5")], operation="sync", first_line=11)
+    assert _jobs(database) == [
+        ("assembling", ["user"]),
+        ("pending", ["user", "assistant"] * 5),
+    ]
+
+
+def test_a_reread_increment_creates_no_second_job(tmp_path):
+    database, job_store, queue, runner = _sdk_runner(tmp_path)
+
+    _read(runner, job_store, _turns(6), operation="sync")
+    _read(runner, job_store, _turns(6), operation="sync")
+    queue.flush_idle_assemblies(now=datetime.now(UTC) + timedelta(minutes=11))
+    _read(runner, job_store, _turns(6), operation="sync")
+    queue.flush_idle_assemblies(now=datetime.now(UTC) + timedelta(minutes=11))
+
+    assert _jobs(database) == [
+        ("pending", ["user", "assistant"] * 5),
+        ("pending", ["user", "assistant"]),
+    ]
+
+
+def test_history_records_with_no_user_message_are_dropped_at_eof(tmp_path):
+    database, job_store, _queue, runner = _sdk_runner(tmp_path)
+
+    _read(
+        runner,
+        job_store,
+        [said("t1", "tool", "ls"), said("a1", "assistant", "ok")],
+        operation="history",
+    )
+
+    assert _jobs(database) == []

memhub/apps/core/tests/test_live_capture.py

@@ -5,6 +5,7 @@ import json
 import os
 import sqlite3
 from contextlib import closing
+from datetime import UTC, datetime, timedelta
 from io import StringIO
 from pathlib import Path
 
@@ -107,7 +108,15 @@ def test_same_message_id_from_the_file_does_not_duplicate(tmp_path) -> None:
     assert store.count() == 1
 
 
+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"
 
@@ -132,10 +141,13 @@ def test_hook_submits_complete_turn_to_the_live_durable_lane(tmp_path) -> None:
         database,
     )
 
+    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:v1:%'"
+            "WHERE idempotency_key LIKE 'turn:v2:%'"
         ).fetchone()
     assert row[0] == "live"
     assert len(json.loads(row[1])) == 2
@@ -161,6 +173,41 @@ def test_hook_submits_complete_turn_to_the_live_durable_lane(tmp_path) -> None:
         assert db.execute("SELECT COUNT(*) FROM evidence").fetchone() == (2,)
 
 
+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 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,
+        )
+
+    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",)]
+
+    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),
+        ("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"
@@ -211,6 +258,7 @@ def test_hook_job_completes_after_file_identity_adoption(tmp_path) -> None:
         },
         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, "我会记住你喜欢蓝莓"),

memhub/apps/core/tests/test_memory_additions.py

@@ -166,6 +166,57 @@ def test_revoked_manual_source_is_not_silently_reauthorized(tmp_path):
     asyncio.run(scenario())
 
 
+@pytest.mark.fresh_sdk
+def test_sdk_manual_add_without_a_grant_row_is_claimed(tmp_path, monkeypatch):
+    """The SDK add path writes no grant row; the claim gate used to require one.
+
+    The intake admitted the job, the drain skipped it on every tick, and the
+    note stayed Pending with attempts=0 while later jobs ran. A missing grant
+    row is allowed; only an explicit revocation holds the job.
+    """
+
+    async def scenario():
+        app = application(tmp_path)
+        async with app.router.lifespan_context(app), client_for(app) as client:
+            addition = {"request_id": str(uuid4()), "text": "Remember this.", "scope": "all-apps"}
+            assert (await client.post("/api/v2/memories", json=addition)).status_code == 201
+            database = app.state.settings.data_dir / "one-memory.sqlite3"
+            with sqlite3.connect(database) as db:
+                assert (
+                    db.execute(
+                        "SELECT COUNT(*) FROM source_grants WHERE source_id='one-memory-manual'"
+                    ).fetchone()[0]
+                    == 0
+                )
+                request_id = db.execute("SELECT request_id FROM memory_note_requests").fetchone()[0]
+            job = app.state.memory_write_queue.get(request_id)
+            assert job is not None and job.state == "pending"
+            monkeypatch.setattr(
+                app.state.model_profile_store, "active_identity", lambda: ("profile", 1)
+            )
+
+            assert app.state.memory_write_drain.claim(limit=1) == (job.job_id,)
+
+            # An explicit revocation still holds the job and refuses new adds.
+            app.state.memory_write_queue.settle(job.job_id, state="pending", owner=None)
+            with sqlite3.connect(database) as db:
+                db.execute(
+                    "INSERT INTO source_grants (user_id,source_id,root_path,requested_path,"
+                    "root_identity,candidate_kind,granted_at,revoked_at,revision) VALUES "
+                    "('local','one-memory-manual','x','x','manual','default',"
+                    "'2026-10-01T00:00:00+00:00','2026-10-27T00:00:00+00:00',1)"
+                )
+            assert app.state.memory_write_drain.claim(limit=1) == ()
+            refused = await client.post(
+                "/api/v2/memories",
+                json={**addition, "request_id": str(uuid4())},
+            )
+            assert refused.status_code == 409
+            assert refused.json()["detail"] == "memory_add_source_revoked"
+
+    asyncio.run(scenario())
+
+
 @pytest.mark.fresh_sdk
 def test_sdk_memory_write_accepts_openai_messages_without_evidence_tables(tmp_path):
     async def scenario():

memhub/apps/core/tests/test_memory_library_all_scenes.py

@@ -90,6 +90,7 @@ class FakeSdk:
         self.productivity = {}
         self.search_calls = []
         self.list_calls = []
+        self.stats_calls = []
         self.get_calls = []
         self.update_calls = []
         self.delete_calls = []
@@ -227,6 +228,22 @@ class FakeSdk:
             "elapsed_ms": 1.0,
         }
 
+    def memory_stats(self, *, user_id, statuses=None):
+        self.stats_calls.append({"user_id": user_id, "statuses": statuses})
+        experiences = [
+            variant for (kind, variant, _id) in self.productivity if kind == "experience"
+        ]
+        facts = [variant for (kind, variant, _id) in self.productivity if kind == "fact"]
+        scenes = {"chat": {"count": len(self.chat)}}
+        for variant in ("coding", "working"):
+            scenes[variant] = {"count": experiences.count(variant), "facts": facts.count(variant)}
+        return {
+            "total": len(self.chat) + len(experiences),
+            "scenes": scenes,
+            "request_id": "stats",
+            "elapsed_ms": 1.0,
+        }
+
     def get(self, memory_id, kind=None, variant=None):
         self.get_calls.append((memory_id, kind, variant))
         match = self._matches(memory_id, kind, variant)
@@ -647,3 +664,33 @@ def test_scene_filter_narrows_search_and_binds_the_window(tmp_path, monkeypatch)
         assert len(sdk.search_calls) == 2
 
     run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_home_summary_counts_what_the_memory_page_lists(tmp_path, monkeypatch):
+    """Home showed 55 (chat only) while the memory page showed 94."""
+
+    async def scenario(app, sdk, client):
+        sdk.add_chat(chat_row("chat-1", "cursor chat"))
+        sdk.add_chat(chat_row("chat-2", "codebuddy chat", source="codebuddy"))
+        sdk.add_productivity("experience", experience_row("exp-w", "task w", "solution w"))
+        sdk.add_productivity(
+            "experience", experience_row("exp-c", "task c", "solution c", variant="coding")
+        )
+        # Facts are not listed on the memory page, so they are not counted.
+        sdk.add_productivity("fact", fact_row("fact-1", "The pytest suite runs in CI"))
+
+        response = await client.get("/api/v2/home-summary")
+        assert response.status_code == 200
+        body = response.json()
+        library = (await client.get("/api/v2/memory-library")).json()
+        assert body["memory_count"] == library["total"] == 4
+        assert sdk.stats_calls == [{"user_id": "local", "statuses": ["active"]}]
+        # The newest memory across scenes, not the newest chat memory.
+        home_list = sdk.list_calls[0]
+        assert home_list["scene"] == "all"
+        assert (home_list["offset"], home_list["limit"], home_list["order"]) == (0, 1, "desc")
+        assert body["latest"]["memory_id"] == "exp-w"
+        assert body["latest"]["scene"] == "working"
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)

memhub/apps/core/tests/test_memory_server_durable_queue.py

@@ -1118,6 +1118,75 @@ def test_sdk_write_processor_uses_add_extracted_for_memory_content_and_additions
     assert len(client.calls) == 1
 
 
+def test_a_batch_with_no_user_or_assistant_message_is_nothing_to_remember(tmp_path) -> None:
+    """Tool-only input came back 400 "content or messages is required" (16 unknown).
+
+    The SDK strips tool records on the chat path. Such a batch completes with
+    no memories; the SDK is never called and nothing is recorded unknown.
+    """
+    from one_memory.imports.models import WrittenBy
+    from one_memory.sdk_write_processor import SdkWriteProcessor
+
+    client = _SdkPublicClient()
+    processor = SdkWriteProcessor(_SdkPublicHandle(client))  # type: ignore[arg-type]
+    tools = tuple(
+        _sdk_evidence(f"tool-{index}", text="3 passed").model_copy(
+            update={"role": "tool", "written_by": WrittenBy.TOOL}
+        )
+        for index in range(2)
+    )
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    record = queue.submit(
+        WriteRequest(
+            internal_session_id="sess-tools",
+            source_id="cursor",
+            batch_identity=BatchIdentity(
+                internal_session_id="sess-tools", first_round=1, last_round=1
+            ),
+            evidence_ids=[item.evidence_id for item in tools],
+            input_records=tools,
+            idempotency_key="tools-only",
+        )
+    )
+    _claim_through_drain(queue, record.job_id)
+
+    asyncio.run(
+        execute_job(
+            record.job_id,
+            queue=queue,
+            supervisor=processor,
+            evidence=_EvidenceStore(set()),
+        )
+    )
+
+    assert client.calls == []
+    assert client.events == []
+    [row] = _rows(path)
+    assert (row["state"], row["hy_invoked"], row["sdk_memory_ids_json"]) == (COMPLETED, 0, "[]")
+
+
+def test_the_drain_tick_writes_a_quiet_live_conversation(tmp_path, monkeypatch) -> None:
+    """The 10-minute idle flush runs on the drain's maintenance timer."""
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    flushed: list[str] = []
+    monkeypatch.setattr(queue, "flush_idle_assemblies", lambda: flushed.append("flush") or ())
+    drain = DurableWriteDrain(queue, execute=_never_execute, reap_interval=timedelta(0))
+
+    async def tick() -> None:
+        task = asyncio.create_task(drain.run_forever())
+        for _ in range(100):
+            await asyncio.sleep(0.01)
+            if flushed:
+                break
+        drain.stop()
+        await task
+
+    asyncio.run(tick())
+    assert flushed
+
+
 def test_sdk_unknown_outcome_inquiry_uses_only_public_receipt() -> None:
     from one_memory.sdk_write_processor import SdkWriteProcessor, SdkWriteReceipt
 

memhub/apps/core/tests/test_policy_api_wiring.py

@@ -127,6 +127,28 @@ def test_the_manual_source_is_admitted_by_grant_without_a_toggle(tmp_path) -> No
     assert _sedimentation_claim_enabled(app, "one-memory-manual") is False
 
 
+def test_the_manual_source_is_allowed_without_a_grant_row_until_revoked(tmp_path) -> None:
+    """The SDK add path writes no grant row, and intake admits the record.
+
+    Claim must answer the same: a missing row is allowed, an explicit
+    revocation is not.
+    """
+    receipt: list[object] = [None]
+    app = SimpleNamespace(
+        state=SimpleNamespace(
+            config_store=ConfigStore(tmp_path),
+            model_profile_store=SimpleNamespace(active_identity=lambda: ("profile", 1)),
+            source_grant_store=SimpleNamespace(receipt=lambda **_kwargs: receipt[0]),
+        )
+    )
+
+    assert _source_writes_on(app, "one-memory-manual") is True
+    assert _sedimentation_claim_enabled(app, "one-memory-manual") is True
+    receipt[0] = SimpleNamespace(revoked_at="2026-10-27")
+    assert _source_writes_on(app, "one-memory-manual") is False
+    assert _sedimentation_claim_enabled(app, "one-memory-manual") is False
+
+
 def test_claim_gate_waits_without_polling_when_no_model_is_active(tmp_path) -> None:
     config = ConfigStore(tmp_path)
     app = SimpleNamespace(

memhub/apps/core/tests/test_sdk_data_plane_migration.py

@@ -705,6 +705,29 @@ def test_cutover_waits_until_the_new_runtime_reads_every_chat_memory(store, acti
     assert store.authority() is MemoryAuthority.SDK
 
 
+def test_sdk_migrate_repairs_metadata_after_cutover_and_writes_the_receipt(store, active_profile):
+    """76/76 memories showed source=default after an upgrade whose repair failed.
+
+    The repair now runs here, offline, right after the cutover commits; Core's
+    startup finds the receipt and does nothing.
+    """
+    _confirmed(store, "m1")
+    home = store.database_path.parent
+
+    result = run_startup_migration(
+        data_dir=home,
+        runtime_build="21",
+        sdk_version="2.0.5",
+        namespace="governed-v2",
+        open_client=_reading({"m1"}),
+    )
+
+    assert result.state == "cutover"
+    receipt = home / f".sdk-metadata-repair-v2-{result.migration_id}.json"
+    assert receipt.is_file()
+    assert json.loads(receipt.read_text())["migration_id"] == result.migration_id
+
+
 def test_retired_coding_experiences_do_not_block_but_an_empty_store_does(store, active_profile):
     _confirmed(store, "chat-1")
     _confirmed(store, "exp-kept", "exp-retired", kind="coding")

memhub/apps/core/tests/test_unified_write_entry_wiring.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
+import json
 import sqlite3
 from uuid import uuid4
 
@@ -197,17 +198,20 @@ def test_live_import_callback_submits_realtime_priority(tmp_path) -> None:
             )
             app.state.config_store.set_sync(source_id, True)
             user = _committed_record(app, source_id=source_id, message_id="live-user")
-            record = app.state.evidence_store.put(
-                EvidenceInput(
-                    source_id=source_id,
-                    conversation_id=user.conversation_id,
-                    message_id="live-assistant",
-                    role="assistant",
-                    written_by=WrittenBy.AGENT,
-                    verbatim="Live reply",
-                    origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
-                )
-            ).record
+            # Live writes by the turn rule: 5 complete turns, the fifth
+            # completed by the next user message.
+            for index in range(10):
+                record = app.state.evidence_store.put(
+                    EvidenceInput(
+                        source_id=source_id,
+                        conversation_id=user.conversation_id,
+                        message_id=f"live-{index}",
+                        role="assistant" if index % 2 == 0 else "user",
+                        written_by=WrittenBy.AGENT if index % 2 == 0 else WrittenBy.PERSON,
+                        verbatim=f"Live line {index}",
+                        origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+                    )
+                ).record
             callback = app.state.import_supervisor.on_stable_evidence
             assert callback is not None
             token = live_sync_arrival.set(True)
@@ -216,11 +220,12 @@ def test_live_import_callback_submits_realtime_priority(tmp_path) -> None:
             finally:
                 live_sync_arrival.reset(token)
             with sqlite3.connect(app.state.evidence_store.database_path) as db:
-                priority = db.execute(
-                    "SELECT priority FROM memory_write_jobs WHERE idempotency_key LIKE ?",
+                rows = db.execute(
+                    "SELECT priority,evidence_ids_json FROM memory_write_jobs "
+                    "WHERE idempotency_key LIKE ?",
                     ("turn:v1:%",),
-                ).fetchone()[0]
-            assert priority == "live"
+                ).fetchall()
+            assert [(priority, len(json.loads(ids))) for priority, ids in rows] == [("live", 10)]
 
     asyncio.run(scenario())
 

memhub/apps/core/tests/test_whole_turn_hy_input.py

@@ -1,8 +1,8 @@
-"""D74: a turn reaches HY whole; only what HY will read counts toward the ceiling."""
+"""D74: a turn reaches HY whole; only what HY will read counts toward the batch size."""
 
 from datetime import UTC, datetime
 
-from one_memory.conversation_processing import _conversation_chunks
+from one_memory.conversation_processing import turn_batches
 from one_memory.imports.models import EvidenceInput, EvidenceOrigin, OriginMetadata, WrittenBy
 from one_memory.imports.storage import EvidenceStore
 
@@ -36,66 +36,33 @@ def test_tool_heavy_turn_is_one_window(tmp_path):
         *[("assistant", "定位中") for _ in range(30)],
         ("assistant", "原因是 fixture 没清理" * 100),
     )
-    chunks = _conversation_chunks(records)
-    assert len(chunks) == 1
-    assert chunks[0] == tuple(records)
+    batches = turn_batches(records, flush=True).batches
+    assert batches == (tuple(records),)
 
 
 def test_long_answer_is_not_cut(tmp_path):
     records = _turn(tmp_path, ("user", "写一份方案"), ("assistant", "段落。" * 7_000))
-    chunks = _conversation_chunks(records)
-    assert chunks == [tuple(records)]
-    assert all(":segment:" not in record.evidence_id for record in chunks[0])
+    batches = turn_batches(records, flush=True).batches
+    assert batches == (tuple(records),)
+    assert all(":segment:" not in record.evidence_id for record in batches[0])
 
 
-def test_text_over_the_ceiling_still_splits(tmp_path):
-    records = _turn(tmp_path, ("user", "整理"), ("assistant", "句子。" * 25_000))
-    chunks = _conversation_chunks(records)
-    assert len(chunks) > 1
-    assert all(chunk[0].evidence_id == records[0].evidence_id for chunk in chunks)
+def test_a_turn_far_over_the_batch_size_is_still_one_window(tmp_path):
+    """The rule never splits a turn: it is written alone, whole.
 
-
-def test_switch_restores_the_old_split(tmp_path, monkeypatch):
-    monkeypatch.setenv("ONE_MEMORY_SPLIT_TURNS", "1")
-    records = _turn(tmp_path, ("user", "写一份方案"), ("assistant", "段落。" * 7_000))
-    assert len(_conversation_chunks(records)) > 1
-
-
-def test_explicit_limits_keep_old_semantics(tmp_path):
-    records = _turn(tmp_path, ("user", "q"), *[("assistant", "a" * 100) for _ in range(30)])
-    assert len(_conversation_chunks(records, max_turns=6, max_characters=1200)) > 1
-
-
-def test_one_window_is_capped_in_raw_bytes(tmp_path):
-    """HY counts 2 KiB per tool record, but we still hold the raw text.
-
-    `client.add` is in-process, so a window's raw bytes are what the process
-    actually holds even though HY will cut each tool record down before it
-    reaches the model. Past the byte ceiling the window is cut, so no single
-    call can grow without bound.
+    The old windows cut text past 60k characters and raw bytes past 2 MB into
+    segments that repeated the user question. HY reads at most
+    `_HY_TOOL_CHARACTERS` of each tool record, so the window HY reads stays
+    bounded by the turn's own text.
     """
-    from one_memory.conversation_processing import _WINDOW_BYTE_CEILING
-
     records = _turn(
         tmp_path,
         ("user", "跑一下全部测试"),
         *[("tool", "y" * 400_000) for _ in range(20)],
+        ("assistant", "句子。" * 25_000),
     )
-    chunks = _conversation_chunks(records)
-    assert len(chunks) > 1
-    assert all(
-        sum(len(item.verbatim.encode("utf-8")) for item in chunk) <= _WINDOW_BYTE_CEILING
-        for chunk in chunks
-    )
-
-
-def test_the_byte_ceiling_counts_bytes_not_characters(tmp_path):
-    """One CJK character is three bytes; a character count would let three
-    times as much through."""
-    from one_memory.conversation_processing import _WINDOW_BYTE_CEILING, _window_bytes
-
-    assert _window_bytes(_turn(tmp_path, ("user", "中文"))[0]) > len("中文")
-    assert _WINDOW_BYTE_CEILING == 2_000_000
+    batches = turn_batches(records, flush=True).batches
+    assert batches == (tuple(records),)
 
 
 def test_the_sdk_strips_tool_messages_on_the_chat_path():

memhub/package.json

@@ -2,7 +2,7 @@
   "name": "one-memory",
   "version": "1.0.5",
   "release": {
-    "build_number": 32,
+    "build_number": 33,
     "macos_min_version": "13.0",
     "app_update": "manual",
     "runtime_manifest_schema": 1,