refactor(memhub): simplify durable memory scheduling

Co-authored-by: Cursor <cursoragent@cursor.com>

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

@@ -133,7 +133,7 @@ from one_memory.imports.jobs import (
 from one_memory.imports.live_sync import LiveSyncHealth, PollingLiveSyncSupervisor
 from one_memory.imports.live_watch import WATCH_MODE, FileWatchLiveSync
 from one_memory.imports.locations import SourceLocation, inspect_source_location
-from one_memory.imports.models import EvidenceOrigin, EvidenceRecord, WrittenBy
+from one_memory.imports.models import EvidenceOrigin, EvidenceRecord
 from one_memory.imports.runner import live_sync_arrival
 from one_memory.imports.sources.common import SourceInstallState
 from one_memory.imports.sources.registry import (
@@ -1159,9 +1159,7 @@ def create_app(
                 # write gate admits it by its grant. Every manual note's job was
                 # therefore skipped on every drain tick — the note stayed
                 # "saved, waiting" with no task behind it, forever.
-                source_claim_enabled=lambda source_id: _sedimentation_claim_enabled(
-                    app, source_id
-                ),
+                source_claim_enabled=lambda source_id: _sedimentation_claim_enabled(app, source_id),
             )
             app.state.memory_write_drain = memory_write_drain
             app.state.memory_write_drain_task = watched_task(
@@ -1190,10 +1188,7 @@ def create_app(
                 for chunk in _conversation_chunks(conversation, max_turns=12, completed_only=True):
                     if (
                         record not in chunk
-                        or not any(
-                            item.role == "user" and item.written_by.value == "person"
-                            for item in chunk
-                        )
+                        or not any(item.role == "user" for item in chunk)
                         or not any(item.role == "assistant" for item in chunk)
                     ):
                         continue
@@ -3690,9 +3685,7 @@ def create_app(
         async with app.state.sedimentation_lock:
             previous = store.sedimentation_enabled()
             if previous is not request.enabled:
-                await asyncio.to_thread(
-                    store.set_sedimentation_enabled, request.enabled
-                )
+                await asyncio.to_thread(store.set_sedimentation_enabled, request.enabled)
                 changed = True
                 diagnostic_event(
                     level="info",
@@ -4068,6 +4061,7 @@ async def _authorize_answer_generation(
     # The explicit question is this operation's input, not a standing permission for history.
     if not question.strip() or not grounding:
         raise MemoryAnswerError("memory_answer_generation_not_authorized")
+
     def check() -> None:
         if _model_profile_store(app).active_identity() != (
             profile.profile_id,
@@ -4091,6 +4085,7 @@ async def _authorize_answer_generation(
     await recheck()
     return recheck
 
+
 class _LazyASGIApp:
     """Compatibility ASGI app that is built only when a server imports it.
 
@@ -4473,7 +4468,7 @@ async def _has_large_old_history(adapter: SourceAdapter, *, cutoff: datetime) ->
             if observed.tzinfo is None:
                 observed = observed.replace(tzinfo=UTC)
             has_old_data = has_old_data or observed.astimezone(UTC) < cutoff
-        if event.evidence.role == "user" and event.evidence.written_by is WrittenBy.PERSON:
+        if event.evidence.role == "user":
             user_turns += 1
         if user_turns > ONBOARDING_LARGE_HISTORY_USER_TURNS and has_old_data:
             return True
@@ -5095,10 +5090,7 @@ def _history_import_gate_released(app: FastAPI) -> bool:
         for job in jobs.list_jobs()
         if job.operation == "history"
         and job.source_id in selected
-        and (
-            state.started_at is None
-            or job.created_at >= state.started_at
-        )
+        and (state.started_at is None or job.created_at >= state.started_at)
     ]
     if {job.source_id for job in matching} != selected:
         return False

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

@@ -196,7 +196,7 @@ def read_raw_context(
         if _prefers_declared_facts(request.query)
         else "1"
     )
-    person_authored = "CASE WHEN e.written_by='person' THEN 1 ELSE 0 END"
+    person_authored = "CASE WHEN e.role='user' THEN 1 ELSE 0 END"
     query_echo = (
         "CASE WHEN trim(e.verbatim)=? OR trim(e.verbatim)=? THEN 0 ELSE 1 END"
         if _prefers_declared_facts(request.query)
@@ -470,9 +470,8 @@ def revalidate_raw_context(
                     )
                 except (ValueError, TypeError):
                     break
-                if (row["role"], row["written_by"], created) != (
+                if (row["role"], created) != (
                     message.role,
-                    message.written_by,
                     message.created_at,
                 ):
                     break

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

@@ -66,12 +66,12 @@ def _field(text: str) -> ResumeField | None:
     return None
 
 
-def _raw_class(written_by: str, evidence_id: str, accepted: frozenset[str]) -> EvidenceClass:
-    if written_by == "person":
+def _raw_class(role: str, evidence_id: str, accepted: frozenset[str]) -> EvidenceClass:
+    if role == "user":
         return "person_statement"
-    if written_by == "agent":
+    if role == "assistant":
         return "agent_report"
-    if written_by == "tool" and evidence_id in accepted:
+    if role == "tool" and evidence_id in accepted:
         return "system_acceptance"
     return "tool_output"
 
@@ -118,7 +118,7 @@ def assemble_resume_packet(
                     field=_field(message.text) or "unclassified",
                     text=message.text,
                     evidence_class=_raw_class(
-                        message.written_by,
+                        message.role,
                         message.evidence_id,
                         verified_system_evidence_ids,
                     ),

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

@@ -345,7 +345,7 @@ def _lexical_raw_rows(
         if context_raw._prefers_declared_facts(query.query)
         else "1"
     )
-    person_authored = "CASE WHEN e.written_by='person' THEN 1 ELSE 0 END"
+    person_authored = "CASE WHEN e.role='user' THEN 1 ELSE 0 END"
     query_echo = (
         "CASE WHEN trim(e.verbatim)=? OR trim(e.verbatim)=? THEN 0 ELSE 1 END"
         if context_raw._prefers_declared_facts(query.query)
@@ -507,7 +507,7 @@ def _raw_item(row: sqlite3.Row, start1: int, matched_by: tuple[str, ...]) -> Raw
             "source_id": row["source_id"],
             "conversation_id": row["conversation_id"],
             "role": row["role"],
-            "written_by": "person" if row["written_by"] == "person" else "agent",
+            "written_by": "person" if row["role"] == "user" else "agent",
             "event_time": event_time.isoformat() if event_time else None,
             "spans": [{"start_cp": start, "end_cp": start + len(text), "text": text}],
             "matched_by": list(matched_by),

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

@@ -449,9 +449,7 @@ class HyMemoryBridge:
         guards = getattr(client, "_one_memory_write_guards", ())
         from one_memory.hy_response_guard import HyReconcileGuard
 
-        reconcile_guards = tuple(
-            guard for guard in guards if isinstance(guard, HyReconcileGuard)
-        )
+        reconcile_guards = tuple(guard for guard in guards if isinstance(guard, HyReconcileGuard))
         for guard in reconcile_guards:
             guard.begin_write()
         before = [len(guard.failures) for guard in guards]
@@ -1129,7 +1127,7 @@ class HyMemoryBridge:
                                 self.profile,
                                 api_key=self.api_key,
                                 data_dir=self.data_dir,
-                        http_transport_factory=self._http_transport_factory,
+                                http_transport_factory=self._http_transport_factory,
                             )
 
                     await _joined_thread(initialize)
@@ -1347,9 +1345,7 @@ async def _with_native_mutation_delta(
     """
 
     retired = set(result.retired_memory_ids)
-    superseded = {
-        key: tuple(values) for key, values in result.superseded_memory_ids.items()
-    }
+    superseded = {key: tuple(values) for key, values in result.superseded_memory_ids.items()}
     chat_targets = {
         key
         for guard in reconcile_guards
@@ -1357,7 +1353,8 @@ async def _with_native_mutation_delta(
         if isinstance(key, str) and key
     }
 
-    for item in raw.get("ops", ()) if isinstance(raw.get("ops"), list) else ():
+    operations = raw.get("ops")
+    for item in operations if isinstance(operations, list) else ():
         if not isinstance(item, Mapping):
             continue
         if item.get("action") == "DELETE":
@@ -1365,7 +1362,8 @@ async def _with_native_mutation_delta(
             if isinstance(key, str) and key:
                 retired.add(key)
 
-    for item in raw.get("fact_ops", ()) if isinstance(raw.get("fact_ops"), list) else ():
+    fact_operations = raw.get("fact_ops")
+    for item in fact_operations if isinstance(fact_operations, list) else ():
         if not isinstance(item, Mapping) or item.get("action") != "SUPERSEDE":
             continue
         old, new = item.get("fact_id"), item.get("superseded_by")
@@ -1386,9 +1384,7 @@ async def _with_native_mutation_delta(
         if isinstance(replacement, str) and replacement:
             superseded[key] = (replacement,)
         elif isinstance(replacement, list):
-            values = tuple(
-                value for value in replacement if isinstance(value, str) and value
-            )
+            values = tuple(value for value in replacement if isinstance(value, str) and value)
             if values:
                 superseded[key] = values
 

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

@@ -7,7 +7,7 @@ from typing import Any, Literal
 
 from pydantic import BaseModel, ConfigDict
 
-from one_memory.imports.models import EvidenceContentKind, EvidenceRecord, WrittenBy
+from one_memory.imports.models import EvidenceContentKind, EvidenceRecord
 from one_memory.input_sanitization import sanitize_model_input
 
 
@@ -61,25 +61,7 @@ class HyWriteBindings:
                     seed.workspace,
                 )
                 or record.content_kind is not EvidenceContentKind.RAW_CONVERSATION
-                or (record.role, record.written_by)
-                not in {
-                    # D98: the same change as the turn assembly. A sub agent's
-                    # question is a `user` message marked written by the
-                    # agent; excluding it here would make the source context
-                    # disagree with what was actually sent.
-                    ("user", WrittenBy.PERSON),
-                    # A host that leaves the author blank still sent a turn.
-                    # WorkBuddy records 1078 of them across 1002 conversations,
-                    # and refusing the write over that answered every one of
-                    # those tasks with `model_content_outcome_unknown` on the
-                    # second attempt: a memory that may have been written and
-                    # can never be confirmed. The author is missing, not the
-                    # turn; scope is still checked above.
-                    ("user", WrittenBy.UNKNOWN),
-                    ("user", WrittenBy.AGENT),
-                    ("assistant", WrittenBy.AGENT),
-                    ("tool", WrittenBy.TOOL),
-                }
+                or record.role not in {"user", "assistant", "tool"}
             ):
                 raise ValueError("hy_memory_write_verification_context_scope")
         self.all_sources: list[dict[str, Any]] = [

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

@@ -79,6 +79,8 @@ class EvidenceInput(BaseModel):
     source_ordinal: str | None = Field(default=None, min_length=1, max_length=256)
     content_kind: EvidenceContentKind = EvidenceContentKind.RAW_CONVERSATION
     role: str = Field(min_length=1, max_length=128)
+    # Legacy import/storage metadata only. Protocol `role` is authoritative;
+    # this value must not affect identity, admission, scheduling or ownership.
     written_by: WrittenBy
     created_at: datetime | None = None
     verbatim: str = Field(min_length=1, max_length=MAX_VERBATIM_CHARACTERS)
@@ -149,17 +151,22 @@ class EvidenceInput(BaseModel):
                 )
             )
         else:
+            # `role` is the protocol-level speaker identity. `written_by` was a
+            # parser guess and differed between two imports of the same host
+            # message (notably sub-agent/user and unknown/user), minting two
+            # Evidence rows. Keep it out of every new identity.
+            #
             # No workspace and no source_ordinal in the derived key: the hook
             # lane has neither (it reports no workspace and a constant ordinal),
             # so including either gave the same message a different key per lane
             # and stored it twice. Both are record metadata here, not identity.
             identity = "\0".join(
                 (
-                    "derived-v3",
+                    "derived-v4",
                     self.user_id,
                     self.source_id,
                     self.conversation_id,
-                    self.written_by,
+                    self.role,
                     self.content_hash,
                 )
             )
@@ -168,6 +175,54 @@ class EvidenceInput(BaseModel):
             # two rows. The capture time is record metadata, not identity.
         return hashlib.sha256(identity.encode()).hexdigest()
 
+    @property
+    def written_by_derived_v3_idempotency_key(self) -> str:
+        """The last identity that treated the parser's author guess as canonical.
+
+        New builds key by protocol role. During an upgrade this key lets a
+        rescan find and reuse the row already stored by derived-v3 instead of
+        inserting the same host message under a second Evidence id.
+        """
+        identity = "\0".join(
+            (
+                "derived-v3",
+                self.user_id,
+                self.source_id,
+                self.conversation_id,
+                self.written_by,
+                self.content_hash,
+            )
+        )
+        return hashlib.sha256(identity.encode()).hexdigest()
+
+    @property
+    def written_by_derived_v3_idempotency_keys(self) -> tuple[str, ...]:
+        """Every legacy author guess this protocol role may have carried."""
+        guesses = {
+            "user": (
+                WrittenBy.PERSON,
+                WrittenBy.AGENT,
+                WrittenBy.UNKNOWN,
+            ),
+            "assistant": (WrittenBy.AGENT, WrittenBy.UNKNOWN, WrittenBy.PERSON),
+            "tool": (WrittenBy.TOOL, WrittenBy.UNKNOWN),
+        }.get(self.role, tuple(WrittenBy))
+        ordered = (self.written_by, *guesses)
+        keys: list[str] = []
+        for written_by in dict.fromkeys(ordered):
+            identity = "\0".join(
+                (
+                    "derived-v3",
+                    self.user_id,
+                    self.source_id,
+                    self.conversation_id,
+                    written_by,
+                    self.content_hash,
+                )
+            )
+            keys.append(hashlib.sha256(identity.encode()).hexdigest())
+        return tuple(keys)
+
     @property
     def legacy_message_idempotency_key(self) -> str | None:
         """The pre-workspace key for a message-shaped input, when one exists.
@@ -212,7 +267,13 @@ class EvidenceInput(BaseModel):
         """
         return self.derived_v2_idempotency_key("derived-v2", workspace_agnostic=False)
 
-    def derived_v2_idempotency_key(self, label: str, *, workspace_agnostic: bool) -> str:
+    def derived_v2_idempotency_key(
+        self,
+        label: str,
+        *,
+        workspace_agnostic: bool,
+        written_by: WrittenBy | None = None,
+    ) -> str:
         identity = "\0".join(
             (
                 label,
@@ -220,7 +281,7 @@ class EvidenceInput(BaseModel):
                 self.source_id,
                 *([] if workspace_agnostic else [self.workspace or "\uffffnull"]),
                 self.conversation_id,
-                self.written_by,
+                written_by or self.written_by,
                 self.created_at.isoformat() if self.created_at else "",
                 self.source_ordinal or "",
                 # v2 hashed the cleaned body; v1 predates raw_verbatim entirely,
@@ -230,6 +291,26 @@ class EvidenceInput(BaseModel):
         )
         return hashlib.sha256(identity.encode()).hexdigest()
 
+    @property
+    def older_written_by_idempotency_keys(self) -> tuple[str, ...]:
+        keys: list[str] = []
+        for written_by in WrittenBy:
+            keys.append(
+                self.derived_v2_idempotency_key(
+                    "derived-v1",
+                    workspace_agnostic=True,
+                    written_by=written_by,
+                )
+            )
+            keys.append(
+                self.derived_v2_idempotency_key(
+                    "derived-v2",
+                    workspace_agnostic=False,
+                    written_by=written_by,
+                )
+            )
+        return tuple(keys)
+
     @property
     def evidence_id(self) -> str:
         return str(uuid5(EVIDENCE_NAMESPACE, self.idempotency_key))
@@ -249,6 +330,7 @@ class EvidenceRecord(BaseModel):
     source_ordinal: str | None
     content_kind: EvidenceContentKind
     role: str
+    # Retained while old databases and export formats still carry the column.
     written_by: WrittenBy
     created_at: datetime | None
     verbatim: str

memhub/apps/core/src/one_memory/imports/sources/codex.py

@@ -141,11 +141,18 @@ def parse_codex_record(
     if role == "user":
         written_by = _user_author(record, payload, session)
         # Machine inputs may quote a whole transcript, including user_query tags.
-        # Session provenance is not proof of who wrote any particular future turn.
-        if written_by is WrittenBy.PERSON:
+        # Protocol role is authoritative; host author metadata does not change
+        # how a user message is cleaned.
+        if not (
+            content.startswith(
+                "The following is the Codex agent history whose request action you are assessing."
+            )
+            and ">>> TRANSCRIPT START" in content
+            and ">>> TRANSCRIPT END" in content
+        ):
             content = strip_host_wrappers(content)
-            if not content:
-                return ParsedSkip(ScanIssueCode.NO_TEXT)
+        if not content:
+            return ParsedSkip(ScanIssueCode.NO_TEXT)
     return ParsedMessage(
         conversation_id=conversation_from_relative_path(context.relative_path),
         message_id=first_string(payload.get("id"), record.get("id")),

memhub/apps/core/src/one_memory/imports/sources/cursor_unified.py

@@ -421,10 +421,7 @@ class CursorUnifiedSourceAdapter:
             if (
                 isinstance(event, ScannedEvidence)
                 and event.evidence.conversation_id in transcript_ids
-                and not (
-                    event.evidence.written_by is WrittenBy.AGENT
-                    and is_host_generated_user_turn(event.evidence.verbatim)
-                )
+                and not is_host_generated_user_turn(event.evidence.verbatim)
             ):
                 yield ScanSkip(
                     cursor=event.cursor,
@@ -449,10 +446,7 @@ def _shadowed_by_transcript(event: ScanEvent, transcript_ids: set[str]) -> ScanE
     if (
         isinstance(event, ScannedEvidence)
         and event.evidence.conversation_id in transcript_ids
-        and not (
-            event.evidence.written_by is WrittenBy.AGENT
-            and is_host_generated_user_turn(event.evidence.verbatim)
-        )
+        and not is_host_generated_user_turn(event.evidence.verbatim)
     ):
         return ScanSkip(
             cursor=event.cursor,

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

@@ -57,7 +57,8 @@ _UNANSWERED_PREFIXES = (
 
 
 def _is_visible_user_row(role: str, written_by: str, verbatim: str) -> bool:
-    if role != "user" or written_by != WrittenBy.PERSON.value:
+    del written_by  # Legacy storage metadata; protocol role is authoritative.
+    if role != "user":
         return False
     return not verbatim.lstrip().lower().startswith(_UNANSWERED_PREFIXES)
 
@@ -468,16 +469,54 @@ class EvidenceStore:
             if anchored is not None:
                 self._write_origins(connection, anchored, evidence.origin_meta, now)
                 workspace, project = self._resolve_conversation_attribution(connection, evidence)
-                connection.execute(
-                    """
-                    UPDATE evidence
-                    SET workspace = COALESCE(workspace, ?),
-                        project = COALESCE(project, ?),
-                        updated_at = ?
-                    WHERE evidence_id = ?
-                    """,
-                    (workspace, project, now, anchored),
-                )
+                existing = connection.execute(
+                    "SELECT role,revision FROM evidence WHERE evidence_id=?",
+                    (anchored,),
+                ).fetchone()
+                if existing is None:
+                    raise RuntimeError("anchored evidence disappeared")
+                if str(existing["role"]) != evidence.role:
+                    was_revised = True
+                    revision = int(existing["revision"]) + 1
+                    connection.execute(
+                        """
+                        UPDATE evidence
+                        SET workspace=COALESCE(workspace,?),
+                            project=COALESCE(project,?),
+                            content_kind=?,role=?,created_at=COALESCE(?,created_at),
+                            verbatim=?,content_hash=?,revision=?,updated_at=?
+                        WHERE evidence_id=?
+                        """,
+                        (
+                            workspace,
+                            project,
+                            evidence.content_kind,
+                            evidence.role,
+                            evidence.created_at.isoformat() if evidence.created_at else None,
+                            evidence.verbatim,
+                            evidence.content_hash,
+                            revision,
+                            now,
+                            anchored,
+                        ),
+                    )
+                    self._insert_current_version(
+                        connection,
+                        anchored,
+                        revision=revision,
+                        changed_at=now,
+                    )
+                else:
+                    connection.execute(
+                        """
+                        UPDATE evidence
+                        SET workspace=COALESCE(workspace,?),
+                            project=COALESCE(project,?),
+                            updated_at=?
+                        WHERE evidence_id=?
+                        """,
+                        (workspace, project, now, anchored),
+                    )
                 record = self._get_with_connection(connection, anchored)
                 if record is None:
                     raise RuntimeError("anchored evidence disappeared")
@@ -485,7 +524,11 @@ class EvidenceStore:
                     from one_memory import ingest_journal
 
                     ingest_journal.record(connection, anchored, cause="import")
-                return PutEvidenceResult(record=record, was_created=False, was_revised=False)
+                return PutEvidenceResult(
+                    record=record,
+                    was_created=False,
+                    was_revised=was_revised,
+                )
             insert = connection.execute(
                 """
                 INSERT INTO evidence (
@@ -803,27 +846,20 @@ class EvidenceStore:
                 and same_scope[0]["idempotency_key"] != evidence.idempotency_key
             ):
                 return str(same_scope[0]["evidence_id"])
-        elif (legacy_id := claim_legacy_key(evidence.legacy_derived_idempotency_key)) is not None:
-            return legacy_id
-        if not evidence.message_id and not evidence.correlation_id:
-            # BUG-03 moved the derived key off the capture time and onto the
-            # pre-cleaning text. A replay of a row stored under the old scheme
-            # has to be recognized as the same message.
-            replayed = connection.execute(
-                """
-                SELECT evidence_id FROM evidence
-                WHERE idempotency_key = ? AND user_id = ? AND source_id = ?
-                  AND conversation_id = ?
-                """,
+        elif (
+            legacy_id := next(
                 (
-                    evidence.prior_derived_idempotency_key,
-                    evidence.user_id,
-                    evidence.source_id,
-                    evidence.conversation_id,
+                    anchored
+                    for key in (
+                        *evidence.written_by_derived_v3_idempotency_keys,
+                        *evidence.older_written_by_idempotency_keys,
+                    )
+                    if (anchored := claim_legacy_key(key)) is not None
                 ),
-            ).fetchone()
-            if replayed is not None:
-                return str(replayed["evidence_id"])
+                None,
+            )
+        ) is not None:
+            return legacy_id
         if (
             EvidenceOrigin.HOOK in evidence.origins
             and evidence.message_id is None
@@ -838,7 +874,7 @@ class EvidenceStore:
                   ON hook.evidence_id=e.evidence_id AND hook.origin='hook'
                 WHERE e.user_id=? AND e.source_id=? AND e.conversation_id=?
                   AND (e.workspace IS ? OR e.workspace IS NULL OR ? IS NULL)
-                  AND e.role=? AND e.written_by=? AND e.content_hash=?
+                  AND e.role=? AND e.content_hash=?
                   AND e.message_id IS NOT NULL AND hook.evidence_id IS NULL
                   AND ABS((julianday(e.created_at)-julianday(?))*86400.0) <= 5.0
                 ORDER BY ABS((julianday(e.created_at)-julianday(?))*86400.0), e.evidence_id
@@ -851,7 +887,6 @@ class EvidenceStore:
                     evidence.workspace,
                     evidence.workspace,
                     evidence.role,
-                    evidence.written_by,
                     evidence.content_hash,
                     evidence.created_at.isoformat(),
                     evidence.created_at.isoformat(),
@@ -872,7 +907,7 @@ class EvidenceStore:
               AND (e.workspace IS ? OR e.workspace IS NULL OR ? IS NULL)
               -- Unary plus keeps the coverage index from looking better than
               -- the message index, which narrows to this exact text.
-              AND e.role = ? AND +e.written_by = ? AND e.content_hash = ?
+              AND e.role = ? AND e.content_hash = ?
               AND e.message_id IS NULL
             """,
             (
@@ -882,7 +917,6 @@ class EvidenceStore:
                 evidence.workspace,
                 evidence.workspace,
                 evidence.role,
-                evidence.written_by,
                 evidence.content_hash,
             ),
         ).fetchall()
@@ -1247,7 +1281,7 @@ class EvidenceStore:
             rows = connection.execute(
                 """
                 SELECT evidence_id FROM evidence
-                WHERE user_id = 'local' AND role = 'user' AND written_by = 'person'
+                WHERE user_id = 'local' AND role = 'user'
                 ORDER BY imported_at, evidence_id
                 LIMIT ?
                 """,
@@ -1301,7 +1335,7 @@ class EvidenceStore:
             rows = connection.execute(
                 f"""
                 SELECT evidence_id FROM evidence
-                WHERE user_id = 'local' AND role = 'user' AND written_by = 'person'
+                WHERE user_id = 'local' AND role = 'user'
                   AND ({predicates})
                 ORDER BY imported_at, evidence_id
                 LIMIT ?
@@ -1333,7 +1367,7 @@ class EvidenceStore:
             rows = connection.execute(
                 f"""
                 SELECT evidence_id FROM evidence
-                WHERE user_id = 'local' AND role = 'user' AND written_by = 'person'
+                WHERE user_id = 'local' AND role = 'user'
                   AND workspace IS NOT NULL
                   AND ({predicates})
                 ORDER BY imported_at, evidence_id
@@ -1476,7 +1510,7 @@ class EvidenceStore:
         )
         values: list[object] = [user_id, *(source_ids or ())]
         visible_user = """
-            role='user' AND written_by='person' AND NOT (
+            role='user' AND NOT (
               lower(ltrim(verbatim)) LIKE '<codex_internal_context%' OR
               lower(ltrim(verbatim)) LIKE '<environment_context>%' OR
               lower(ltrim(verbatim)) LIKE '<heartbeat>%' OR
@@ -1485,7 +1519,7 @@ class EvidenceStore:
             )
         """
         selected_visible_user = """
-            e.role='user' AND e.written_by='person' AND NOT (
+            e.role='user' AND NOT (
               lower(ltrim(e.verbatim)) LIKE '<codex_internal_context%' OR
               lower(ltrim(e.verbatim)) LIKE '<environment_context>%' OR
               lower(ltrim(e.verbatim)) LIKE '<heartbeat>%' OR
@@ -1638,7 +1672,7 @@ class EvidenceStore:
         if seed is None:
             return ConversationTurnPage((), {}, 0, offset, limit)
         visible_user = """
-          role='user' AND written_by='person' AND NOT (
+          role='user' AND NOT (
             lower(ltrim(verbatim)) LIKE '<codex_internal_context%' OR
             lower(ltrim(verbatim)) LIKE '<environment_context>%' OR
             lower(ltrim(verbatim)) LIKE '<heartbeat>%' OR
@@ -1731,7 +1765,7 @@ class EvidenceStore:
                       AND first.conversation_id = evidence.conversation_id
                       AND first.workspace IS evidence.workspace
                       AND (
-                        (first.role = 'user' AND first.written_by = 'person')
+                        first.role = 'user'
                         OR first.content_kind = 'memory_content'
                       )
                     ORDER BY julianday(COALESCE(first.created_at, first.imported_at)),
@@ -2125,8 +2159,6 @@ class EvidenceStore:
             (
                 existing["content_kind"] != evidence.content_kind,
                 existing["role"] != evidence.role,
-                evidence.written_by is not WrittenBy.UNKNOWN
-                and existing["written_by"] != evidence.written_by,
                 existing["content_hash"] != evidence.content_hash,
             )
         )

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

@@ -85,15 +85,14 @@ def _state(connection: sqlite3.Connection, evidence_id: str) -> dict[str, str |
     value = evidence.model_dump(mode="json")
     # The stored normalized content_hash is insufficient for exact-content leases.
     content = _digest(value["verbatim"])
-    material = _digest(
-        {key: value[key] for key in ("verbatim", "role", "written_by", "content_kind")}
-    )
+    material = _digest({key: value[key] for key in ("verbatim", "role", "content_kind")})
     origins = _digest(value["origin_meta"])
     identity = _digest(
         {
             key: item
             for key, item in value.items()
-            if key not in {"verbatim", "origin_meta", "imported_at", "updated_at"}
+            if key
+            not in {"verbatim", "written_by", "origin_meta", "imported_at", "updated_at"}
         }
     )
     return dict(

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

@@ -210,7 +210,8 @@ def record_capture_rejection(
         payload: dict[str, object] = {"source_id": source_id, "reason": reason.value}
         if marker.exists() and now - marker.stat().st_mtime < window_seconds:
             payload = json.loads(marker.read_text(encoding="utf8"))
-            payload["count"] = int(payload.get("count", 1)) + 1
+            previous = payload.get("count", 1)
+            payload["count"] = int(previous) + 1 if isinstance(previous, (str, int, float)) else 2
         else:
             payload["count"] = 1
             payload["window_started_at"] = datetime.fromtimestamp(now, UTC).isoformat()
@@ -235,7 +236,7 @@ def _submit_live_memory(store: EvidenceStore, record: EvidenceRecord, *, changed
     ):
         if (
             record in chunk
-            and any(item.role == "user" and item.written_by is WrittenBy.PERSON for item 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")

memhub/apps/core/src/one_memory/matters/discovery.py

@@ -160,7 +160,9 @@ class MatterDiscovery:
         task_by_id: dict[str, MemoryTask] = {}
         for chunk in chunks:
             for record in chunk:
-                task = await asyncio.to_thread(processor.tasks.task_for_evidence, record.evidence_id)
+                task = await asyncio.to_thread(
+                    processor.tasks.task_for_evidence, record.evidence_id
+                )
                 if task is not None and task.status is MemoryTaskStatus.COMPLETED:
                     task_by_id[task.task_id] = task
         tasks = tuple(sorted(task_by_id.values(), key=lambda task: (task.created_at, task.task_id)))

memhub/apps/core/src/one_memory/matters/synthesis.py

@@ -270,7 +270,7 @@ async def reconcile_current(
         (
             r
             for r in records
-            if r.written_by == "person"
+            if r.role == "user"
             and r.created_at
             and (r.source_id, r.conversation_id) in related
         ),

memhub/apps/core/src/one_memory/matters/validation.py

@@ -51,14 +51,19 @@ def validate_claim(claim: Claim, records: Mapping[str, EvidenceRecord | SampleRe
         text = record.text if isinstance(record, SampleRecord) else record.verbatim
         if not citation.quote.strip() or citation.quote not in text:
             raise ValueError("citation quote is not in the source")
-        authors.add(str(record.written_by))
+        author = {
+            "user": "person",
+            "assistant": "agent",
+            "tool": "tool",
+        }.get(record.role, "unknown")
+        authors.add(author)
         created = record.created_at
         citations.append(
             citation.model_copy(
                 update={
                     "revision": record.revision,
                     "source_id": record.source_id,
-                    "written_by": str(record.written_by),
+                    "written_by": author,
                     "created_at": created.isoformat() if isinstance(created, datetime) else created,
                 }
             )

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

@@ -290,7 +290,7 @@ def read_memory_library(
                            COALESCE(
                                MIN(
                                    CASE WHEN e.role = 'user'
-                                         AND e.written_by IN ('person', 'user')
+                                         AND e.role = 'user'
                                         THEN e.created_at END
                                ),
                                MIN(e.created_at)

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

@@ -53,7 +53,7 @@ from one_memory.hy_memory_bridge import (
 from one_memory.hy_runtime_config import hy_memory_version
 from one_memory.hy_source_context import HySourceContext, HySupportedMemory
 from one_memory.imports.access import opaque_workspace_id
-from one_memory.imports.models import EvidenceContentKind, EvidenceOrigin, EvidenceRecord, WrittenBy
+from one_memory.imports.models import EvidenceContentKind, EvidenceOrigin, EvidenceRecord
 from one_memory.imports.runner import live_sync_arrival
 from one_memory.imports.storage import EvidenceStore
 from one_memory.input_sanitization import sanitize_model_input
@@ -187,7 +187,6 @@ UNIFIED_HY_AGENT_ID = "one-memory"
 PRE_SEND_REFUSAL_CODES = frozenset(
     {
         "hy_memory_input_exceeds_window",
-        "hy_memory_write_verification_context_scope",
     }
 )
 _BACKFILL_BATCH_SIZE = 8
@@ -297,6 +296,7 @@ class MemoryTaskRunner:
         *,
         timeout: float | None = None,
         before_hy_invocation: Callable[[], Awaitable[None]] | None = None,
+        input_records: tuple[EvidenceRecord, ...] | None = None,
     ) -> MemoryTask:
         started = monotonic()
         diagnostic_event(
@@ -307,13 +307,28 @@ class MemoryTaskRunner:
             task_id=task_id,
         )
         try:
-            body = (
-                self._run_without_content_permission(task_id)
-                if before_hy_invocation is None
-                else self._run_without_content_permission(
-                    task_id, before_hy_invocation=before_hy_invocation
+            if input_records is None:
+                body = (
+                    self._run_without_content_permission(task_id)
+                    if before_hy_invocation is None
+                    else self._run_without_content_permission(
+                        task_id,
+                        before_hy_invocation=before_hy_invocation,
+                    )
+                )
+            else:
+                body = (
+                    self._run_without_content_permission(
+                        task_id,
+                        input_records=input_records,
+                    )
+                    if before_hy_invocation is None
+                    else self._run_without_content_permission(
+                        task_id,
+                        before_hy_invocation=before_hy_invocation,
+                        input_records=input_records,
+                    )
                 )
-            )
             # The ceiling bounds how long this task holds its harness's slot:
             # a harness is serial, so one hung task starves every turn behind
             # it. It is insurance against a single bad call, not a way to run
@@ -398,6 +413,7 @@ class MemoryTaskRunner:
         task_id: str,
         *,
         before_hy_invocation: Callable[[], Awaitable[None]] | None = None,
+        input_records: tuple[EvidenceRecord, ...] | None = None,
     ) -> MemoryTask:
         # Every store call here is synchronous, and several are writes that
         # wait on the write lock for up to five seconds. Made from the loop,
@@ -451,7 +467,9 @@ class MemoryTaskRunner:
         step_started = monotonic()
         frozen = await in_the_background(self.tasks.inputs, task.task_id)
         chunk: tuple[EvidenceRecord, ...] | None
-        if frozen:
+        if input_records is not None:
+            chunk = input_records
+        elif frozen:
             try:
                 chunk = await in_the_background(self._frozen_chunk, task)
             except HyMemoryBridgeError as error:
@@ -484,8 +502,11 @@ class MemoryTaskRunner:
                     self._source_context, task, request.source_context
                 )
                 request = request.model_copy(update={"source_context": context})
+            uses_frozen_input = bool(frozen) or input_records is not None
             grant_token = (
-                await in_the_background(self._grant_token, task, chunk[0]) if frozen else None
+                await in_the_background(self._grant_token, task, chunk[0])
+                if uses_frozen_input
+                else None
             )
             memory_contents = tuple(
                 sanitize_model_input(record.verbatim)
@@ -528,19 +549,24 @@ class MemoryTaskRunner:
                     agent_id=request.agent_id,
                 ),
             )
-            if frozen:
-                await in_the_background(self._frozen_chunk, task)
+            if uses_frozen_input:
+                if input_records is None:
+                    await in_the_background(self._frozen_chunk, task)
                 if grant_token != await in_the_background(self._grant_token, task, chunk[0]):
                     raise HyMemoryBridgeError("hy_memory_source_changed")
             if request.source_context is not None:
-                await in_the_background(self._check_source_context, request.source_context)
+                await in_the_background(
+                    self._check_source_context,
+                    request.source_context,
+                    skip_current=input_records is not None,
+                )
             # One thread hop for the whole batch rather than a dozen writes
             # made from the loop.
             step_started = monotonic()
             await in_the_background(
                 self._record_outputs,
                 task=task,
-                chunk=self._original_records(chunk),
+                chunk=chunk,
                 outputs=outputs,
                 request=request,
                 derivation=derivation,
@@ -778,8 +804,18 @@ class MemoryTaskRunner:
                 raise HyMemoryBridgeError("hy_memory_evidence_changed") from error
         return context
 
-    def _check_source_context(self, context: HySourceContext) -> None:
-        for original in (*context.current, *context.history, *context.supporting):
+    def _check_source_context(
+        self,
+        context: HySourceContext,
+        *,
+        skip_current: bool = False,
+    ) -> None:
+        checked = (
+            (*context.history, *context.supporting)
+            if skip_current
+            else (*context.current, *context.history, *context.supporting)
+        )
+        for original in checked:
             record = self.evidence.get(original.evidence_id)
             if record is None or not matches_evidence_ref(record, evidence_ref(original)):
                 raise HyMemoryBridgeError("hy_memory_evidence_changed")
@@ -880,7 +916,7 @@ class MemoryTaskRunner:
             # HY's own attribution. An undecided memory counts as the person's:
             # HY's prompt defaults that way.
             from_agent = output.owner == "agent" or all(
-                record.written_by is WrittenBy.AGENT
+                record.role == "assistant"
                 and record.content_kind is EvidenceContentKind.RAW_CONVERSATION
                 for record in support
             )
@@ -1063,11 +1099,7 @@ class MemoryTaskRunner:
             record.content_kind is EvidenceContentKind.RAW_CONVERSATION for record in chunk
         )
         source = next(
-            (
-                record
-                for record in chunk
-                if record.role == "user" and record.written_by is WrittenBy.PERSON
-            ),
+            (record for record in chunk if record.role == "user"),
             chunk[0],
         )
         request_turns = tuple(
@@ -1076,12 +1108,9 @@ class MemoryTaskRunner:
             if (
                 record.content_kind is EvidenceContentKind.RAW_CONVERSATION
                 and (
-                    # D98: any `user` message goes, whoever the parser says
-                    # wrote it. A sub agent's question is ordinary
-                    # user-assistant data; the author field cannot tell who
-                    # wrote a message, and filtering on it dropped the sub
-                    # agent's turns. `written_by` still labels the display and
-                    # decides owner — it no longer decides what is sent.
+                    # Protocol role is authoritative. A sub-agent question is
+                    # an ordinary user message regardless of a host parser's
+                    # former author guess.
                     record.role == "user"
                     # The assistant's side goes too. Without it 「就按你说的那个来」
                     # arrives with nothing to resolve it against, and HY's own
@@ -1092,7 +1121,7 @@ class MemoryTaskRunner:
                     or record.role == "assistant"
                     # Tool calls go as they happened; whether they make the
                     # turn a coding turn is HY's call, not ours.
-                    or (record.role == "tool" and record.written_by is WrittenBy.TOOL)
+                    or record.role == "tool"
                 )
             )
             or (not has_raw and record.content_kind is EvidenceContentKind.MEMORY_CONTENT)
@@ -1376,7 +1405,6 @@ class MemoryProcessingSupervisor:
         async with self._explicit_processing_lock(records):
             if self._sedimentation_stopped():
                 raise HyMemoryBridgeError("model_sedimentation_disabled")
-            records = await in_the_background(self._original_records, records)
             active = await in_the_background(self.profiles.active, self.secrets)
             if active is None or (profile_id is not None and profile_id != active.profile_id):
                 raise HyMemoryBridgeError("model_profile_not_active")
@@ -1391,13 +1419,6 @@ class MemoryProcessingSupervisor:
                 raise HyMemoryBridgeError("hy_memory_source_disabled")
             conversations: dict[tuple[str, str, str | None], list[EvidenceRecord]] = {}
             for record in records:
-                stored = await in_the_background(self.evidence.get, record.evidence_id)
-                if (
-                    stored is None
-                    or stored.revision != record.revision
-                    or stored.content_hash != record.content_hash
-                ):
-                    raise ValueError("hy_memory_evidence_changed")
                 conversations.setdefault(
                     (record.source_id, record.conversation_id, record.workspace), []
                 ).append(record)
@@ -1414,7 +1435,7 @@ class MemoryProcessingSupervisor:
                 for conversation in conversations.values()
                 for chunk in _conversation_chunks(conversation, max_turns=12, completed_only=True)
             ]
-            prepared = []
+            prepared: list[tuple[MemoryTask, tuple[EvidenceRecord, ...]]] = []
             for chunk in chunks:
                 task = await in_the_background(
                     self._enqueue_chunk,
@@ -1422,12 +1443,12 @@ class MemoryProcessingSupervisor:
                     active,
                     freeze_inputs=True,
                 )
-                prepared.append(task)
+                prepared.append((task, chunk))
             runner = self.runner_factory()
             completed = []
             if on_progress:
-                on_progress(tuple(prepared), 0)
-            for task in prepared:
+                on_progress(tuple(task for task, _chunk in prepared), 0)
+            for task, chunk in prepared:
                 if task.status is not MemoryTaskStatus.COMPLETED:
                     # Another importer may already own this task. Never race its claim.
                     task = await in_the_background(self.tasks.require, task.task_id)
@@ -1456,23 +1477,41 @@ class MemoryProcessingSupervisor:
                                 work_id=task.task_id,
                                 scope_key=chunk[0].source_id,
                             ):
-                                task = (
-                                    await runner.run(task.task_id, timeout=timeout)
-                                    if before_hy_invocation is None
-                                    else await runner.run(
-                                        task.task_id,
-                                        timeout=timeout,
-                                        before_hy_invocation=before_hy_invocation,
+                                if isinstance(runner, MemoryTaskRunner):
+                                    task = (
+                                        await runner.run(
+                                            task.task_id,
+                                            timeout=timeout,
+                                            input_records=chunk,
+                                        )
+                                        if before_hy_invocation is None
+                                        else await runner.run(
+                                            task.task_id,
+                                            timeout=timeout,
+                                            before_hy_invocation=before_hy_invocation,
+                                            input_records=chunk,
+                                        )
+                                    )
+                                else:
+                                    task = (
+                                        await runner.run(task.task_id, timeout=timeout)
+                                        if before_hy_invocation is None
+                                        else await runner.run(
+                                            task.task_id,
+                                            timeout=timeout,
+                                            before_hy_invocation=before_hy_invocation,
+                                        )
                                     )
-                                )
                         finally:
                             _unmark_direct_task(self.tasks.database_path, task.task_id)
                 if task.status is not MemoryTaskStatus.COMPLETED:
                     raise HyMemoryBridgeError(task.last_error_code or "hy_memory_incomplete")
-                await in_the_background(runner._frozen_chunk, task)
                 completed.append(task)
                 if on_progress:
-                    on_progress(tuple(prepared), len(completed))
+                    on_progress(
+                        tuple(prepared_task for prepared_task, _chunk in prepared),
+                        len(completed),
+                    )
             expected = {record.evidence_id for record in records if _is_hy_turn_record(record)}
             actual = {
                 row["evidence_id"] for task in completed for row in self.tasks.inputs(task.task_id)
@@ -3261,9 +3300,7 @@ class MemoryProcessingSupervisor:
         if not tail:
             return False
         last = tail[-1]
-        return bool(
-            last.role == "user" and last.written_by is WrittenBy.PERSON and _opens_turn(last)
-        )
+        return bool(last.role == "user" and _opens_turn(last))
 
     def _admitted_chunks(
         self,
@@ -3608,7 +3645,6 @@ def _is_memory_note(record: EvidenceRecord) -> bool:
     meta = record.origin_meta.get(EvidenceOrigin.MANUAL_IMPORT)
     return (
         record.source_id == "one-memory-manual"
-        and record.written_by is WrittenBy.PERSON
         and meta is not None
         and meta.raw_meta.get("input_mode") == "direct_memory_note"
     )
@@ -3903,14 +3939,12 @@ def _merge_historical_chunks(
 
 
 def _chunk_input_ref(chunk: tuple[EvidenceRecord, ...]) -> str:
-    person_records = tuple(
-        item for item in chunk if item.role == "user" and item.written_by is WrittenBy.PERSON
-    )
-    return person_records[-1].evidence_id if person_records else chunk[-1].evidence_id
+    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
 
 
 def _opens_turn(record: EvidenceRecord) -> bool:
-    return record.role == "user" and record.written_by is WrittenBy.PERSON
+    return record.role == "user"
 
 
 def _trailing_user_turn_is_open(turn: list[EvidenceRecord]) -> bool:
@@ -3927,10 +3961,8 @@ def _trailing_user_turn_is_open(turn: list[EvidenceRecord]) -> bool:
 def _is_hy_turn_record(record: EvidenceRecord) -> bool:
     """What HY receives of a turn: the input turns and the tool calls made in it."""
     if record.role == "tool":
-        return (
-            record.written_by is WrittenBy.TOOL
-            and record.content_kind is EvidenceContentKind.RAW_CONVERSATION
-            and bool(record.verbatim.strip())
+        return record.content_kind is EvidenceContentKind.RAW_CONVERSATION and bool(
+            record.verbatim.strip()
         )
     return _is_hy_input_turn(record)
 
@@ -3947,11 +3979,8 @@ def _is_hy_input_turn(record: EvidenceRecord) -> bool:
     # from a model.
     if record.role == "assistant":
         return True
-    # D98: the author no longer decides. A sub agent's question arrives as a
-    # `user` message marked written by the agent — it is ordinary
-    # user-assistant data, and dropping it sent Claude Code transcripts to HY
-    # with every question but the first missing. `written_by` labels the
-    # display and decides owner; it cannot tell who actually wrote a message.
+    # A sub-agent question is still a protocol `user` message. Host-specific
+    # author guesses do not decide whether it reaches HY.
     if record.role != "user":
         return False
     if source.casefold().startswith(_HARNESS_PREFIXES):

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

@@ -27,7 +27,9 @@ from datetime import datetime
 from enum import StrEnum
 from typing import Annotated, Final, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
+
+from one_memory.imports.models import EvidenceRecord
 
 #: Bumped when a derivation rule changes, so old and new ids never compare equal.
 SCHEMA_VERSION: Final[str] = "ms1"
@@ -167,7 +169,6 @@ class CanonicalMessage(BaseModel):
     message_id: str | None = Field(default=None, min_length=1, max_length=512)
     ordinal: int = Field(ge=0)
     role: str = Field(min_length=1, max_length=128)
-    written_by: str = Field(min_length=1, max_length=64)
     content: str = Field(min_length=1)
     workspace: str | None = Field(default=None, max_length=1024)
 
@@ -214,12 +215,13 @@ class WriteRequest(BaseModel):
     source_id: str | None = Field(default=None, min_length=1, max_length=128)
     batch_identity: BatchIdentity
     evidence_ids: list[str] = Field(min_length=1)
+    # New jobs freeze the complete input in their own payload. The default is
+    # solely for reading pre-upgrade rows whose payload only carried ids.
+    input_records: tuple[EvidenceRecord, ...] = ()
     idempotency_key: str = Field(min_length=1, max_length=256)
     priority: Literal["live", "normal"] = "normal"
     event_at: datetime | None = None
-    time_quality: Literal["exact", "anchored", "session", "approximate"] = (
-        "approximate"
-    )
+    time_quality: Literal["exact", "anchored", "session", "approximate"] = "approximate"
 
     @field_validator("evidence_ids")
     @classmethod
@@ -232,6 +234,16 @@ class WriteRequest(BaseModel):
             raise ValueError("evidence_ids must be unique")
         return value
 
+    @field_validator("input_records")
+    @classmethod
+    def _input_records_match_ids(
+        cls, value: tuple[EvidenceRecord, ...], info: ValidationInfo
+    ) -> tuple[EvidenceRecord, ...]:
+        evidence_ids = info.data.get("evidence_ids")
+        if value and evidence_ids != [record.evidence_id for record in value]:
+            raise ValueError("input_records must match evidence_ids in order")
+        return value
+
     @property
     def request_key(self) -> str:
         """Stable dedup key. Same batch + same intent must not enqueue twice."""

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

@@ -30,7 +30,7 @@ from uuid import uuid4
 from one_memory.background_work import in_the_background
 from one_memory.database_access import main_database
 from one_memory.diagnostic_log import diagnostic_event
-from one_memory.imports.models import EvidenceOrigin, EvidenceRecord, WrittenBy
+from one_memory.imports.models import EvidenceOrigin, EvidenceRecord
 from one_memory.imports.storage import EvidenceStore
 from one_memory.memory_server.contracts import WriteRequest
 from one_memory.memory_server.data_safety import MIN_FREE_BYTES  # noqa: F401 - re-exported
@@ -250,16 +250,9 @@ MODEL_NOT_READY_CODES = frozenset(
 #: `failed`, not `unknown` (which would wrongly imply a possibly-applied
 #: write) and not a retry (which would fail identically).
 #:
-#: `hy_memory_write_verification_context_scope` belongs here for the same
-#: reason: `HyWriteBindings` is built inside `ingest` *before* the payload
-#: reaches HY, and it is the only thing that raises this code. The durable
-#: marker is already set by then, so classifying it as `unknown` claimed a
-#: write may have landed when HY was never called — a sticky row that blocked
-#: the queue behind it and left no reason to act on. See issue #44.
 PRE_SEND_REFUSAL_CODES = frozenset(
     {
         "hy_memory_input_exceeds_window",
-        "hy_memory_write_verification_context_scope",
     }
 )
 
@@ -290,11 +283,6 @@ _VALID_JOB_STATES = frozenset(
 #: A row in one of these states can never be moved back to a live one.
 _TERMINAL_WRITE_STATES = frozenset({COMPLETED, FAILED, DEAD, CANCELLED})
 
-#: A job in one of these states can never run again, so its payload is dead
-#: weight. `unknown` is deliberately absent: it is non-terminal and still needs
-#: its intent for reconciliation. `running` is obviously still live.
-TERMINAL_PAYLOAD_STATES = (COMPLETED, FAILED, DEAD, CANCELLED)
-
 #: Columns a current schema declares but an older `memory_write_jobs` may lack.
 #: `CREATE TABLE IF NOT EXISTS` is a no-op for an existing table, so without
 #: this an upgraded install would keep an older queue that every claim/lease
@@ -404,6 +392,42 @@ def _matches_payload_except_priority(stored: str, payload: str) -> bool:
     return old == new
 
 
+def _matches_legacy_reference_replay(stored: str, payload: str) -> bool:
+    """Match an old id-only caller to the same job's frozen payload."""
+    try:
+        old = json.loads(stored)
+        new = json.loads(payload)
+    except (TypeError, ValueError):
+        return False
+    if not isinstance(old, dict) or not isinstance(new, dict):
+        return False
+    old_inputs = old.get("input_records")
+    new_inputs = new.get("input_records")
+    if bool(old_inputs) == bool(new_inputs):
+        return False
+    old.pop("input_records", None)
+    new.pop("input_records", None)
+    return old == new
+
+
+def _matches_reaped_legacy_payload(stored: str, payload: str) -> bool:
+    """Compare a current frozen request with a pre-snapshot reaped payload."""
+    try:
+        old = json.loads(stored)
+        current = json.loads(payload)
+    except (TypeError, ValueError):
+        return False
+    if (
+        not isinstance(old, dict)
+        or not isinstance(current, dict)
+        or not isinstance(old.get(REAPED_PAYLOAD_MARKER), str)
+    ):
+        return False
+    current.pop("input_records", None)
+    legacy_payload = json.dumps(current, sort_keys=True)
+    return bool(old[REAPED_PAYLOAD_MARKER] == _payload_digest(legacy_payload))
+
+
 class DurableQueueSchemaUnavailableError(RuntimeError):
     """The durable queue schema has not been created yet.
 
@@ -456,6 +480,7 @@ class DurableWriteQueue:
         if not initialize:
             return
         with main_database(database_path).write() as connection, connection:
+            connection.row_factory = sqlite3.Row
             # First create the base table if absent. Then widen an older table
             # before replaying indexes that reference the new columns; a bare
             # CREATE INDEX would fail on a pre-priority queue.
@@ -491,8 +516,7 @@ class DurableWriteQueue:
             "UPDATE memory_write_jobs SET source_key=session_key WHERE source_key=''"
         )
         legacy_rows = connection.execute(
-            "SELECT job_id FROM memory_write_jobs WHERE queue_seq=0 "
-            "ORDER BY created_at, rowid"
+            "SELECT job_id FROM memory_write_jobs WHERE queue_seq=0 ORDER BY created_at, rowid"
         ).fetchall()
         next_seq = int(
             connection.execute(
@@ -539,10 +563,9 @@ class DurableWriteQueue:
             )
             return
         evidence_columns = {
-            str(row[1])
-            for row in connection.execute("PRAGMA table_info(evidence)").fetchall()
+            str(row[1]) for row in connection.execute("PRAGMA table_info(evidence)").fetchall()
         }
-        if not {"created_at", "imported_at", "role", "written_by"} <= evidence_columns:
+        if not {"created_at", "imported_at", "role"} <= evidence_columns:
             connection.execute(
                 "UPDATE memory_write_jobs SET event_at=created_at WHERE event_at IS NULL"
             )
@@ -556,7 +579,7 @@ class DurableWriteQueue:
                     FROM memory_write_job_evidence AS linked
                     JOIN evidence AS e ON e.evidence_id = linked.evidence_id
                     WHERE linked.job_id = memory_write_jobs.job_id
-                      AND e.role = 'user' AND e.written_by = 'person'
+                      AND e.role = 'user'
                 ),
                 (
                     SELECT MIN(COALESCE(e.created_at, e.imported_at))
@@ -666,6 +689,7 @@ class DurableWriteQueue:
         replacing the earlier intent.
         """
         self._refuse_if_below_reserve()
+        request = self._freeze_request_inputs(request)
         payload = json.dumps(
             request.model_dump(
                 mode="json",
@@ -691,16 +715,22 @@ class DurableWriteQueue:
                 (key,),
             ).fetchone()
             if existing is not None:
-                # A replay of work that already finished is a hit, not a
-                # conflict. Terminal rows store a reaped digest instead of the
-                # original payload, and the caller's key is derived from the
-                # evidence's revision, which can move on between replays while
-                # the payload's evidence ids stay the same. Refusing that pair
-                # turned every startup replay of one settled job into an
-                # exception: 236 of them in a single afternoon, all one key.
                 already_settled = (
                     str(existing["state"]) == COMPLETED and int(existing["hy_invoked"]) == 1
                 )
+                stored_payload = str(existing["payload_json"])
+                priority_replay = _matches_payload_except_priority(stored_payload, payload)
+                legacy_reaped_match = bool(
+                    already_settled
+                    and _matches_reaped_legacy_payload(stored_payload, payload)
+                )
+                if (
+                    not _matches_stored_payload(stored_payload, payload)
+                    and not priority_replay
+                    and not _matches_legacy_reference_replay(stored_payload, payload)
+                    and not legacy_reaped_match
+                ):
+                    raise ValueError(f"idempotency key {key} already exists with different content")
                 if already_settled:
                     return DurableWriteRecord(
                         job_id=str(existing["job_id"]),
@@ -711,14 +741,6 @@ class DurableWriteQueue:
                         updated_at=str(existing["updated_at"]),
                         detail="already completed; replay treated as a hit",
                     )
-                priority_replay = _matches_payload_except_priority(
-                    str(existing["payload_json"]), payload
-                )
-                if (
-                    not _matches_stored_payload(str(existing["payload_json"]), payload)
-                    and not priority_replay
-                ):
-                    raise ValueError(f"idempotency key {key} already exists with different content")
                 if priority_replay and request.priority == "live":
                     connection.execute(
                         "UPDATE memory_write_jobs SET priority='live', updated_at=? "
@@ -803,9 +825,7 @@ class DurableWriteQueue:
                     0,
                     None,
                     event_at,
-                    request.time_quality
-                    if request.event_at is not None
-                    else resolved_quality,
+                    request.time_quality if request.event_at is not None else resolved_quality,
                     now,
                     now,
                 ),
@@ -835,34 +855,36 @@ class DurableWriteQueue:
             updated_at=now,
         )
 
-    def _event_time_for_request(
-        self, request: WriteRequest
-    ) -> tuple[datetime | None, str]:
-        if self.evidence is None:
-            return None, "approximate"
-        records = [
-            record
-            for evidence_id in request.evidence_ids
-            if (record := self.evidence.get(evidence_id)) is not None
-        ]
+    def _freeze_request_inputs(self, request: WriteRequest) -> WriteRequest:
+        if request.input_records or self.evidence is None:
+            return request
+        records: list[EvidenceRecord] = []
+        for evidence_id in request.evidence_ids:
+            record = self.evidence.get(evidence_id)
+            if record is None:
+                return request
+            records.append(record)
+        return request.model_copy(update={"input_records": tuple(records)})
+
+    def _event_time_for_request(self, request: WriteRequest) -> tuple[datetime | None, str]:
+        records = list(request.input_records)
+        if not records and self.evidence is not None:
+            records = [
+                record
+                for evidence_id in request.evidence_ids
+                if (record := self.evidence.get(evidence_id)) is not None
+            ]
         if not records:
             return None, "approximate"
         opener = next(
-            (
-                record
-                for record in records
-                if record.role == "user"
-                and record.written_by is WrittenBy.PERSON
-            ),
+            (record for record in records if record.role == "user"),
             records[0],
         )
         if opener.created_at is None:
             return opener.imported_at, "approximate"
         origin = opener.origin_meta.get(EvidenceOrigin.FILE)
         basis = origin.raw_meta.get("time_basis") if origin is not None else None
-        turn_event_at = (
-            origin.raw_meta.get("turn_event_at") if origin is not None else None
-        )
+        turn_event_at = origin.raw_meta.get("turn_event_at") if origin is not None else None
         if isinstance(turn_event_at, str):
             try:
                 return datetime.fromisoformat(turn_event_at), "exact"
@@ -981,11 +1003,55 @@ class DurableWriteQueue:
                 raise ReconcileSchemaUnavailableError(
                     "evidence table is unavailable; run migrations before reconcile"
                 )
-            where = (
-                "WHERE evidence.imported_at IS NOT NULL "
-                "AND NOT EXISTS (SELECT 1 FROM memory_write_job_evidence ref "
-                "WHERE ref.evidence_id=evidence.evidence_id)"
+            has_versions = (
+                connection.execute(
+                    "SELECT 1 FROM sqlite_master "
+                    "WHERE type='table' AND name='evidence_versions'"
+                ).fetchone()
+                is not None
             )
+            evidence_columns = {
+                str(row[1])
+                for row in connection.execute("PRAGMA table_info(evidence)").fetchall()
+            }
+            if {"revision", "content_hash", "updated_at"} <= evidence_columns:
+                revision_time = (
+                    "COALESCE((SELECT version.changed_at FROM evidence_versions version "
+                    "WHERE version.evidence_id=evidence.evidence_id "
+                    "AND version.revision=evidence.revision),"
+                    "evidence.updated_at,evidence.imported_at)"
+                    if has_versions
+                    else "COALESCE(evidence.updated_at,evidence.imported_at)"
+                )
+                where = (
+                    "WHERE evidence.imported_at IS NOT NULL "
+                    "AND NOT EXISTS ("
+                    "SELECT 1 FROM memory_write_job_evidence linked "
+                    "JOIN memory_write_jobs job ON job.job_id=linked.job_id, "
+                    "json_each(job.payload_json,'$.input_records') frozen "
+                    "WHERE linked.evidence_id=evidence.evidence_id "
+                    "AND json_extract(frozen.value,'$.evidence_id')=evidence.evidence_id "
+                    "AND json_extract(frozen.value,'$.revision')=evidence.revision "
+                    "AND json_extract(frozen.value,'$.content_hash')=evidence.content_hash"
+                    ") "
+                    # Pre-snapshot builds only recorded an Evidence id. Treat
+                    # that as coverage only when its Job was created after this
+                    # exact revision.
+                    "AND NOT EXISTS ("
+                    "SELECT 1 FROM memory_write_job_evidence legacy "
+                    "JOIN memory_write_jobs legacy_job ON legacy_job.job_id=legacy.job_id "
+                    "WHERE legacy.evidence_id=evidence.evidence_id "
+                    "AND COALESCE(json_array_length("
+                    "json_extract(legacy_job.payload_json,'$.input_records')),0)=0 "
+                    f"AND legacy_job.created_at>=({revision_time}))"
+                )
+            else:
+                # A genuinely old/minimal schema has no revision identity.
+                where = (
+                    "WHERE evidence.imported_at IS NOT NULL "
+                    "AND NOT EXISTS (SELECT 1 FROM memory_write_job_evidence legacy "
+                    "WHERE legacy.evidence_id=evidence.evidence_id)"
+                )
             params: tuple[Any, ...] = ()
             cursor_clause = ""
             if after_cursor is not None:
@@ -1285,45 +1351,10 @@ class DurableWriteQueue:
         return state
 
     def clear_terminal_payloads(self, *, limit: int = 512) -> int:
-        """Replace terminal rows' request bodies with a content digest.
-
-        A completed/failed/dead/cancelled row can never run again, so the full
-        serialized `WriteRequest` is unbounded dead weight. A digest keeps the
-        idempotency contract intact: a replay of the same intent still resolves
-        to the same row, while the same key with different content still
-        conflicts. `unknown` is left alone because it is non-terminal and its
-        intent is needed for reconciliation.
-
-        Bounded by `limit` so a large backlog drains over several calls instead
-        of one long write on the single writer.
-        """
+        """Deprecated compatibility hook; durable inputs are immutable audit data."""
         if type(limit) is not int or limit < 1:
             raise ValueError("limit must be positive")
-        placeholders = ",".join("?" for _ in TERMINAL_PAYLOAD_STATES)
-        with (
-            main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
-            connection,
-        ):
-            rows = connection.execute(
-                "SELECT job_id, payload_json FROM memory_write_jobs "
-                f"WHERE state IN ({placeholders}) "
-                "AND instr(payload_json, ?) = 0 "
-                "LIMIT ?",
-                (*TERMINAL_PAYLOAD_STATES, REAPED_PAYLOAD_MARKER, limit),
-            ).fetchall()
-            if not rows:
-                return 0
-            for row in rows:
-                connection.execute(
-                    "UPDATE memory_write_jobs SET payload_json=? "
-                    f"WHERE job_id=? AND state IN ({placeholders})",
-                    (
-                        _reaped_payload(_payload_digest(str(row["payload_json"]))),
-                        str(row["job_id"]),
-                        *TERMINAL_PAYLOAD_STATES,
-                    ),
-                )
-        return len(rows)
+        return 0
 
     def readmit_parked(self, *, states: tuple[str, ...] = PARKED_STATES) -> int:
         """Move parked jobs back to `pending` for a fresh claim.
@@ -1870,9 +1901,12 @@ class DurableWriteRegistry:
 def evidence_for_request(store: EvidenceStore, request: WriteRequest) -> tuple[EvidenceRecord, ...]:
     """The Evidence a stored write points at, in the request's own order.
 
-    A write may only point at Evidence that still exists; a missing id is a
-    rejection rather than a silent partial send.
+    Current jobs are self-contained: their immutable records are the model
+    input even if a Harness later rewinds or edits its live transcript. The
+    store lookup remains only for payloads written by pre-upgrade builds.
     """
+    if request.input_records:
+        return request.input_records
     records = []
     for evidence_id in request.evidence_ids:
         record = store.get(evidence_id)
@@ -1979,10 +2013,14 @@ async def execute_job(
         # Never persist the raw message: it may be a provider response or the
         # user's own text, and `last_error` is shown over HTTP as `detail`.
         code = safe_error_code(error)
-        if code in {
-            "model_content_source_disabled",
-            "model_sedimentation_disabled",
-        } and not hy_invoked:
+        if (
+            code
+            in {
+                "model_content_source_disabled",
+                "model_sedimentation_disabled",
+            }
+            and not hy_invoked
+        ):
             await in_the_background(queue.release_unstarted, job_id, owner=owner)
             return
         # The task layer classifies these before any bridge is opened, so HY
@@ -1994,14 +2032,8 @@ async def execute_job(
         # Everything else may have crossed into HY, and HY has no idempotency
         # key, so it is `unknown` rather than a blind retry.
         if code in PRE_SEND_REFUSAL_CODES:
-            # Provably before the bridge, even though the marker is already
-            # set: the write bindings are built — and the input window
-            # checked — before the payload reaches HY, and this code is raised
-            # from that construction alone. Nothing was sent, so this is
-            # terminal rather than unknown, and the marker is cleared so
-            # reclaim does not treat a row HY never saw as possibly-sent.
-            # Otherwise a refused source context pinned the queue behind it
-            # with no reason shown and no way forward. See issue #44.
+            # The frozen input window is checked before the payload reaches
+            # HY. This one refusal therefore has a proven pre-send boundary.
             state, invoked_marker = FAILED, False
         elif hy_invoked:
             # Error names do not prove execution phase: permission and
@@ -2103,7 +2135,6 @@ class DurableWriteDrain:
         #: sustained backlog must not keep finished rows' request bodies in the
         #: business database indefinitely. Its own gate so neither interval can
         #: starve the other.
-        self._last_terminal_reap: datetime | None = None
         #: `waiting_for_model` re-admission is also time-based recoverability:
         #: 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.
@@ -2154,7 +2185,7 @@ class DurableWriteDrain:
         now: datetime | None = None,
         budget_ms: int | None = None,
     ) -> tuple[str, ...]:
-        """Claim at most one FIFO head per enabled Harness."""
+        """Claim event-time heads, with one running job per enabled Harness."""
         if type(limit) is not int or limit < 1:
             raise ValueError("limit must be positive")
         now = now or datetime.now(UTC)
@@ -2196,9 +2227,7 @@ class DurableWriteDrain:
                     (PENDING,),
                 ).fetchall()
             enabled_sources = {
-                str(row[0])
-                for row in source_rows
-                if self.source_writes_permitted(str(row[0]))
+                str(row[0]) for row in source_rows if self.source_writes_permitted(str(row[0]))
             }
             if not enabled_sources:
                 return ()
@@ -2218,8 +2247,7 @@ class DurableWriteDrain:
                 source_clause = f"AND candidate.source_key IN ({placeholders}) "
                 source_params = tuple(sorted(enabled_sources))
             window = connection.execute(
-                "SELECT window_anchor,window_end FROM memory_write_schedule_state "
-                "WHERE singleton=1"
+                "SELECT window_anchor,window_end FROM memory_write_schedule_state WHERE singleton=1"
             ).fetchone()
             if window is not None:
                 anchor_text, end_text = str(window[0]), str(window[1])
@@ -2227,14 +2255,12 @@ class DurableWriteDrain:
                 active_source_params: tuple[Any, ...] = ()
                 if enabled_sources is not None:
                     placeholders = ",".join("?" for _ in enabled_sources)
-                    active_source_clause = (
-                        f"AND active.source_key IN ({placeholders}) "
-                    )
+                    active_source_clause = f"AND active.source_key IN ({placeholders}) "
                     active_source_params = tuple(sorted(enabled_sources))
                 # The contract retires a window when every *eligible*
                 # pending/running job in it has settled or parked. A pending row
-                # that FIFO already holds back (an unexpired lease, a busy
-                # Harness, or an earlier unsent sibling) is not eligible, so it
+                # that scheduling already holds back (an unexpired lease, a
+                # busy Harness, or an earlier event in its Session) is not eligible, so it
                 # must not keep the anchor alive: those blockers usually sit
                 # outside this ten-minute range and can never be claimed from
                 # inside it, which pinned the anchor and starved the whole
@@ -2245,18 +2271,18 @@ class DurableWriteDrain:
                     "WHERE active.event_at>=? AND active.event_at<? AND ("
                     "active.state=? OR (active.state=? "
                     "AND (active.lease_expires_at IS NULL "
-                    "OR active.lease_expires_at < ?) "
-                    + active_source_clause
-                    + "AND NOT EXISTS ("
+                    "OR active.lease_expires_at < ?) " + active_source_clause + "AND NOT EXISTS ("
                     "SELECT 1 FROM memory_write_jobs AS held "
                     "WHERE held.source_key=active.source_key AND held.state=? "
                     "AND held.lease_owner IS NOT NULL "
                     "AND held.lease_expires_at>=?) "
                     "AND NOT EXISTS ("
                     "SELECT 1 FROM memory_write_jobs AS earlier "
-                    "WHERE earlier.source_key=active.source_key "
+                    "WHERE earlier.session_key=active.session_key "
                     f"AND earlier.state IN ({blocking}) "
-                    "AND earlier.queue_seq<active.queue_seq) "
+                    "AND (earlier.event_at<active.event_at OR "
+                    "(earlier.event_at=active.event_at "
+                    "AND earlier.queue_seq<active.queue_seq))) "
                     ")) LIMIT 1",
                     (
                         anchor_text,
@@ -2271,9 +2297,7 @@ class DurableWriteDrain:
                     ),
                 ).fetchone()
                 if active is None:
-                    connection.execute(
-                        "DELETE FROM memory_write_schedule_state WHERE singleton=1"
-                    )
+                    connection.execute("DELETE FROM memory_write_schedule_state WHERE singleton=1")
                     window = None
             if window is None:
                 anchor = connection.execute(
@@ -2281,18 +2305,18 @@ class DurableWriteDrain:
                     "FROM memory_write_jobs AS candidate "
                     "WHERE candidate.state=? "
                     "AND (candidate.lease_expires_at IS NULL "
-                    "OR candidate.lease_expires_at < ?) "
-                    + source_clause
-                    + "AND NOT EXISTS ("
+                    "OR candidate.lease_expires_at < ?) " + source_clause + "AND NOT EXISTS ("
                     "SELECT 1 FROM memory_write_jobs AS held "
                     "WHERE held.source_key=candidate.source_key AND held.state=? "
                     "AND held.lease_owner IS NOT NULL "
                     "AND held.lease_expires_at>=?) "
                     "AND NOT EXISTS ("
                     "SELECT 1 FROM memory_write_jobs AS earlier "
-                    "WHERE earlier.source_key=candidate.source_key "
+                    "WHERE earlier.session_key=candidate.session_key "
                     f"AND earlier.state IN ({blocking}) "
-                    "AND earlier.queue_seq<candidate.queue_seq) "
+                    "AND (earlier.event_at<candidate.event_at OR "
+                    "(earlier.event_at=candidate.event_at "
+                    "AND earlier.queue_seq<candidate.queue_seq))) "
                     "ORDER BY candidate.event_at,candidate.queue_seq LIMIT 1",
                     (
                         PENDING,
@@ -2321,17 +2345,19 @@ class DurableWriteDrain:
                 "FROM memory_write_jobs AS candidate "
                 "WHERE candidate.state=? "
                 "AND (candidate.lease_expires_at IS NULL OR candidate.lease_expires_at < ?) "
-                + source_clause +
-                "AND candidate.event_at>=? AND candidate.event_at<? "
+                + source_clause
+                + "AND candidate.event_at>=? AND candidate.event_at<? "
                 "AND NOT EXISTS ("
                 "SELECT 1 FROM memory_write_jobs AS held "
                 "WHERE held.source_key=candidate.source_key AND held.state=? "
                 "AND held.lease_owner IS NOT NULL AND held.lease_expires_at>=?) "
                 "AND NOT EXISTS ("
                 "SELECT 1 FROM memory_write_jobs AS earlier "
-                "WHERE earlier.source_key=candidate.source_key "
+                "WHERE earlier.session_key=candidate.session_key "
                 f"AND earlier.state IN ({blocking}) "
-                "AND earlier.queue_seq<candidate.queue_seq) "
+                "AND (earlier.event_at<candidate.event_at OR "
+                "(earlier.event_at=candidate.event_at "
+                "AND earlier.queue_seq<candidate.queue_seq))) "
                 "ORDER BY candidate.event_at,candidate.queue_seq",
                 (
                     PENDING,
@@ -2345,12 +2371,17 @@ class DurableWriteDrain:
                 ),
             ).fetchall()
             selected: list[str] = []
+            selected_sources: set[str] = set()
             for row in rows:
+                source_key = str(row["source_key"])
+                if source_key in selected_sources:
+                    continue
                 if self.source_claim_enabled is not None and not self.source_claim_enabled(
-                    str(row["source_key"])
+                    source_key
                 ):
                     continue
                 selected.append(str(row["job_id"]))
+                selected_sources.add(source_key)
                 if len(selected) == min(limit, self.max_concurrent_sources):
                     break
             if not selected:
@@ -2579,19 +2610,6 @@ class DurableWriteDrain:
                         self.queue.reclaim_expired,
                         operation_name="reclaim_expired",
                     )
-                # Terminal-payload reaping is a bounded retention write: a
-                # finished row's full request body must not sit in the business
-                # database forever just because the queue never went idle. Its
-                # own interval keeps it off every poll.
-                if (
-                    self._last_terminal_reap is None
-                    or tick_started - self._last_terminal_reap >= self._reap_interval
-                ):
-                    self._last_terminal_reap = tick_started
-                    await self._run_idle_maintenance(
-                        self.queue.clear_terminal_payloads,
-                        operation_name="clear_terminal_payloads",
-                    )
                 # A model outage clears on its own, so `waiting_for_model` jobs
                 # are re-admitted on a timer. That must not depend on idleness: a
                 # queue busy for longer than the wait window would otherwise keep

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

@@ -1525,7 +1525,7 @@ async def _053_memory_write_event_time(cursor: Any) -> None:
                 FROM memory_write_job_evidence AS linked
                 JOIN evidence AS e ON e.evidence_id = linked.evidence_id
                 WHERE linked.job_id = memory_write_jobs.job_id
-                  AND e.role = 'user' AND e.written_by = 'person'
+                  AND e.role = 'user'
             ),
             (
                 SELECT MIN(COALESCE(e.created_at, e.imported_at))
@@ -1571,6 +1571,22 @@ async def _053_memory_write_event_time(cursor: Any) -> None:
         )
 
 
+async def _054_legacy_possibly_sent_tasks(cursor: Any) -> None:
+    """Preserve the old sent marker before removing its retired permission table."""
+    columns = await cursor.columns("memory_task_continuous_permissions")
+    if "task_id" not in columns:
+        return
+    await cursor.execute(
+        """
+        UPDATE memory_tasks
+        SET hy_invocation_count=CASE
+            WHEN hy_invocation_count<1 THEN 1 ELSE hy_invocation_count END
+        WHERE task_id IN (SELECT task_id FROM memory_task_continuous_permissions)
+        """
+    )
+    await cursor.execute("DROP TABLE memory_task_continuous_permissions")
+
+
 def model_profiles_ddl(dialect: str = "mysql") -> str:
     """Model profiles and their version history.
 
@@ -3022,4 +3038,9 @@ MIGRATIONS: tuple[Migration, ...] = (
         _053_memory_write_event_time,
         ("memory_write_schedule_state",),
     ),
+    Migration(
+        54,
+        "legacy_possibly_sent_tasks",
+        _054_legacy_possibly_sent_tasks,
+    ),
 )

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

@@ -107,6 +107,8 @@ def build_request(
     first_round: int,
     last_round: int,
     evidence_ids: list[str],
+    input_records: tuple[EvidenceRecord, ...] = (),
+    source_id: str | None = None,
     priority: Literal["live", "normal"] = "normal",
     idempotency_key: str | None = None,
 ) -> WriteRequest:
@@ -125,6 +127,8 @@ def build_request(
         internal_session_id=internal_session_id,
         batch_identity=batch,
         evidence_ids=evidence_ids,
+        input_records=input_records,
+        source_id=source_id,
         idempotency_key=idempotency_key or batch.batch_id,
         priority=priority,
     )
@@ -271,6 +275,8 @@ def build_request_from_evidence(
         first_round=first_round,
         last_round=last_round,
         evidence_ids=list(ordered),
+        input_records=tuple(ordered.values()),
+        source_id=seed.source_id,
         priority=priority,
         idempotency_key=idempotency_key,
     )

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

@@ -22,7 +22,7 @@ from datetime import datetime
 from hashlib import sha256
 from typing import Literal
 
-from one_memory.imports.models import EvidenceOrigin, EvidenceRecord, WrittenBy
+from one_memory.imports.models import EvidenceOrigin, EvidenceRecord
 from one_memory.memory_server.contracts import (
     BatchIdentity,
     WriteRequest,
@@ -126,6 +126,11 @@ class UnifiedWriteIntake:
     ) -> WriteRequest:
         session_id = internal_session_id_for(record.conversation_id, record.source_id)
         event_at, time_quality = _job_time((record,))
+        request_key = (
+            f"entry:{record.evidence_id}"
+            if record.revision == 1
+            else (f"entry:v2:{record.evidence_id}:{record.revision}:{record.content_hash}")
+        )
         return WriteRequest(
             internal_session_id=session_id,
             source_id=record.source_id,
@@ -135,7 +140,8 @@ class UnifiedWriteIntake:
                 last_round=last_round,
             ),
             evidence_ids=[record.evidence_id],
-            idempotency_key=f"entry:{record.evidence_id}",
+            input_records=(record,),
+            idempotency_key=request_key,
             priority=priority,
             event_at=event_at,
             time_quality=time_quality,
@@ -176,6 +182,7 @@ class UnifiedWriteIntake:
                     last_round=1,
                 ),
                 evidence_ids=[record.evidence_id for record in records],
+                input_records=records,
                 idempotency_key=request_id,
                 priority=priority,
                 event_at=event_at,
@@ -204,20 +211,14 @@ def _job_time(
     """Use the turn opener's wall clock, with an explicit fallback quality."""
 
     opener = next(
-        (
-            record
-            for record in records
-            if record.role == "user" and record.written_by is WrittenBy.PERSON
-        ),
+        (record for record in records if record.role == "user"),
         records[0],
     )
     if opener.created_at is None:
         return opener.imported_at, "approximate"
     file_origin = opener.origin_meta.get(EvidenceOrigin.FILE)
     basis = file_origin.raw_meta.get("time_basis") if file_origin is not None else None
-    turn_event_at = (
-        file_origin.raw_meta.get("turn_event_at") if file_origin is not None else None
-    )
+    turn_event_at = file_origin.raw_meta.get("turn_event_at") if file_origin is not None else None
     if isinstance(turn_event_at, str):
         try:
             return datetime.fromisoformat(turn_event_at), "exact"

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

@@ -17,7 +17,7 @@ from pydantic import BaseModel, ConfigDict, Field
 
 from one_memory.database import take_the_write_lock_first
 from one_memory.database_access import main_database
-from one_memory.imports.models import EvidenceRecord
+from one_memory.imports.models import EvidenceRecord, WrittenBy
 from one_memory.memory_server.migrations import memory_tasks_ddl
 from one_memory.model_profiles import ModelProfile, ModelProfileSnapshot
 
@@ -174,29 +174,60 @@ def evidence_ref(record: EvidenceRecord) -> TaskEvidenceRef:
             "segment_end": int(marker["end"]),
             "segment_hash": str(marker["hash"]),
         }
-    identity = record.model_dump(mode="json", exclude={"origin_meta", "imported_at", "updated_at"})
     return {
         "evidence_id": record.evidence_id,
         "revision": record.revision,
         "content_hash": record.content_hash,
-        "identity_hash": sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest(),
+        "identity_hash": _evidence_identity_hash(record, include_written_by=False),
     }
 
 
+def _evidence_identity_hash(record: EvidenceRecord, *, include_written_by: bool) -> str:
+    excluded = {"origin_meta", "imported_at", "updated_at"}
+    if not include_written_by:
+        excluded.add("written_by")
+    identity = record.model_dump(mode="json", exclude=excluded)
+    return sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest()
+
+
+def _identity_hash_matches(record: EvidenceRecord, expected: object) -> bool:
+    candidates = {_evidence_identity_hash(record, include_written_by=False)}
+    # Tasks created by pre-role-only builds keep their original proof. The
+    # legacy metadata could also have changed in place without a revision, so
+    # accept the old hash under every historical author guess.
+    for written_by in WrittenBy:
+        candidates.add(
+            _evidence_identity_hash(
+                record.model_copy(update={"written_by": written_by}),
+                include_written_by=True,
+            )
+        )
+    return expected in candidates
+
+
 def matches_evidence_ref(record: EvidenceRecord, ref: TaskEvidenceRef) -> bool:
     if "segment_hash" in ref:
         base = evidence_ref(record)
         if any(
             base.get(key) != ref.get(key)
-            for key in ("evidence_id", "revision", "content_hash", "identity_hash")
+            for key in ("evidence_id", "revision", "content_hash")
         ):
             return False
+        if not _identity_hash_matches(record, ref.get("identity_hash")):
+            return False
         start, end = ref.get("segment_start"), ref.get("segment_end")
         if not isinstance(start, int) or not isinstance(end, int) or not (0 <= start < end):
             return False
         return sha256(record.verbatim[start:end].encode()).hexdigest() == ref["segment_hash"]
     current = evidence_ref(record)
-    return all(current.get(key) == value for key, value in ref.items())
+    return all(
+        (
+            _identity_hash_matches(record, value)
+            if key == "identity_hash"
+            else current.get(key) == value
+        )
+        for key, value in ref.items()
+    )
 
 
 def segment_evidence_record(

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

@@ -87,7 +87,7 @@ def build_acceptance_pack(
             """
             SELECT DISTINCT e.evidence_id
             FROM evidence e
-            WHERE e.role = 'user' AND e.written_by = 'person'
+            WHERE e.role = 'user'
               AND LENGTH(e.verbatim) <= 300
               AND (
                 e.verbatim LIKE '%我是一名%' OR e.verbatim LIKE '%我是一个%'
@@ -118,7 +118,7 @@ def build_acceptance_pack(
         for row in connection.execute(
             """
             SELECT evidence_id FROM evidence
-            WHERE role = 'user' AND written_by != 'person'
+            WHERE role != 'user'
             """
         ):
             strata["attribution_edge"].add(row["evidence_id"])
@@ -138,11 +138,7 @@ def build_acceptance_pack(
             quota,
         )
     if len(selected) < sample_size:
-        eligible = (
-            evidence_id
-            for evidence_id, row in evidence.items()
-            if row["role"] == "user" and row["written_by"] == "person"
-        )
+        eligible = (evidence_id for evidence_id, row in evidence.items() if row["role"] == "user")
         _extend_unique(selected, _stable_order(eligible, seed=f"{seed}:fill"), sample_size)
     selected = selected[:sample_size]
     if len(selected) < sample_size:
@@ -163,7 +159,11 @@ def build_acceptance_pack(
                 row_id=f"{index:03d}",
                 evidence_id=evidence_id,
                 source_id=source["source_id"],
-                written_by=source["written_by"],
+                written_by=(
+                    "person"
+                    if source["role"] == "user"
+                    else ("agent" if source["role"] == "assistant" else source["role"])
+                ),
                 character_count=len(source["verbatim"]),
                 content_hash=source["content_hash"],
                 strata=tuple(

memhub/apps/core/src/one_memory/retrieval/content.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Sequence
 
-from one_memory.imports.models import EvidenceContentKind, EvidenceRecord, WrittenBy
+from one_memory.imports.models import EvidenceContentKind, EvidenceRecord
 from one_memory.imports.sources.parsing import is_host_generated_user_turn, strip_host_wrappers
 from one_memory.input_sanitization import sanitize_model_input
 from one_memory.memory_processing import _is_hy_input_turn
@@ -15,10 +15,7 @@ def identity_content_ready(record: EvidenceRecord) -> bool:
     # cleaning requires a source map and remains explicitly uncovered for now.
     return (
         record.content_kind is EvidenceContentKind.RAW_CONVERSATION
-        and (
-            (record.role == "user" and record.written_by is WrittenBy.PERSON)
-            or (record.role == "assistant" and record.written_by is WrittenBy.AGENT)
-        )
+        and record.role in {"user", "assistant"}
         and _is_hy_input_turn(record)
         and not is_host_generated_user_turn(record.verbatim)
         and strip_host_wrappers(record.verbatim) == record.verbatim.strip()

memhub/apps/core/src/one_memory/retrieval/mapped_content.py

@@ -10,7 +10,6 @@ from one_memory.imports.models import (
     MAX_VERBATIM_CHARACTERS,
     EvidenceContentKind,
     EvidenceRecord,
-    WrittenBy,
 )
 from one_memory.imports.sources.parsing import (
     is_host_generated_user_turn,
@@ -164,10 +163,10 @@ def prepare_mapped_content(record: EvidenceRecord) -> PreparedMappedContent:
     The returned text is transient; only metadata is suitable for an outer manifest.
     That manifest must independently bind Evidence identity/revision/hash and grants.
     """
-    if record.content_kind is not EvidenceContentKind.RAW_CONVERSATION or not (
-        (record.role == "user" and record.written_by is WrittenBy.PERSON)
-        or (record.role == "assistant" and record.written_by is WrittenBy.AGENT)
-    ):
+    if record.content_kind is not EvidenceContentKind.RAW_CONVERSATION or record.role not in {
+        "user",
+        "assistant",
+    }:
         raise MappedContentError("mapped_content_ineligible")
     original = record.verbatim
     if len(original) > MAX_VERBATIM_CHARACTERS:

memhub/apps/core/src/one_memory/retrieval/raw_evidence.py

@@ -11,7 +11,7 @@ from datetime import UTC
 
 from one_memory import ingest_journal as journal
 from one_memory.context_raw import _check, _connection
-from one_memory.imports.models import EvidenceRecord, WrittenBy
+from one_memory.imports.models import EvidenceRecord
 from one_memory.imports.storage import EvidenceStore
 from one_memory.retrieval.index_store import RetrievalIndexStore
 from one_memory.retrieval.models import EvidenceSpan, RawEvidence
@@ -166,7 +166,7 @@ def materialize_raw_candidates(
                         source_id=record.source_id,
                         conversation_id=record.conversation_id,
                         role=record.role,
-                        written_by="person" if record.written_by is WrittenBy.PERSON else "agent",
+                        written_by="person" if record.role == "user" else "agent",
                         event_time=event_time,
                         spans=tuple(spans),
                         matched_by=("semantic",),

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

@@ -56,10 +56,10 @@ def build_source_import_receipt(
             "SELECT count(*) AS messages, "
             "count(DISTINCT conversation_id || char(0) || COALESCE(workspace, '')) "
             "AS conversations, "
-            "coalesce(sum(written_by='person'),0) AS person_messages, "
-            "coalesce(sum(written_by='agent'),0) AS agent_messages, "
-            "coalesce(sum(written_by='tool'),0) AS tool_messages, "
-            "coalesce(sum(written_by NOT IN ('person','agent','tool')),0) "
+            "coalesce(sum(role='user'),0) AS person_messages, "
+            "coalesce(sum(role='assistant'),0) AS agent_messages, "
+            "coalesce(sum(role='tool'),0) AS tool_messages, "
+            "coalesce(sum(role NOT IN ('user','assistant','tool')),0) "
             "AS unknown_author_messages, "
             "coalesce(sum(created_at IS NULL),0) AS unknown_time_messages "
             "FROM evidence WHERE user_id=? AND source_id=?",
@@ -68,7 +68,9 @@ def build_source_import_receipt(
         # Bound text reads before materializing snippets. These are samples, not a
         # recency ranking; sorting every large original body is unnecessary here.
         examples = db.execute(
-            "SELECT evidence_id, role, written_by, created_at, "
+            "SELECT evidence_id, role, CASE role "
+            "WHEN 'user' THEN 'person' WHEN 'assistant' THEN 'agent' "
+            "WHEN 'tool' THEN 'tool' ELSE 'unknown' END AS written_by, created_at, "
             "substr(verbatim,1,600) AS excerpt, length(verbatim)>600 AS truncated "
             "FROM evidence WHERE evidence_id IN ("
             "SELECT evidence_id FROM evidence WHERE user_id=? AND source_id=? LIMIT 3)",
@@ -80,10 +82,13 @@ def build_source_import_receipt(
         # Qualify before LIMIT: agent/system rows must not consume the three samples.
         query_filter = " AND evidence_id=?" if query_evidence_id is not None else ""
         queries = db.execute(
-            f"""SELECT evidence_id, revision AS evidence_revision, role, written_by, created_at,
-                CASE WHEN user_id=? AND source_id=? AND role='user' AND written_by='person'
+            f"""SELECT evidence_id, revision AS evidence_revision, role,
+                CASE role WHEN 'user' THEN 'person' WHEN 'assistant' THEN 'agent'
+                    WHEN 'tool' THEN 'tool' ELSE 'unknown' END AS written_by,
+                created_at,
+                CASE WHEN user_id=? AND source_id=? AND role='user'
                     AND length(verbatim)<=65536
-                    THEN display_query_projection(verbatim,role,written_by)
+                    THEN display_query_projection(verbatim,role,'person')
                     ELSE NULL END AS projection
                 FROM evidence WHERE user_id=? AND source_id=?{query_filter}
                     AND projection IS NOT NULL

memhub/apps/core/tests/test_evidence_store.py

@@ -2,6 +2,7 @@ import json
 import sqlite3
 from concurrent.futures import ThreadPoolExecutor
 from datetime import UTC, datetime
+from hashlib import sha256
 
 import pytest
 from pydantic import ValidationError
@@ -14,6 +15,7 @@ from one_memory.imports.models import (
     WrittenBy,
 )
 from one_memory.imports.storage import EvidenceStore
+from one_memory.memory_tasks import evidence_ref, matches_evidence_ref
 from one_memory.workspace_resolution import KnownWorkspace
 
 
@@ -25,6 +27,7 @@ def evidence(
     verbatim: str = "Remember the source, not the guess.",
     origin: EvidenceOrigin = EvidenceOrigin.FILE,
     written_by: WrittenBy = WrittenBy.PERSON,
+    role: str = "user",
     source_ordinal: str | None = None,
 ) -> EvidenceInput:
     return EvidenceInput(
@@ -35,7 +38,7 @@ def evidence(
         source_ordinal=source_ordinal
         if source_ordinal is not None
         else ("line:1" if message_id is None and correlation_id is None else None),
-        role="user",
+        role=role,
         written_by=written_by,
         created_at=datetime(2026, 8, 31, 12, 0, tzinfo=UTC),
         verbatim=verbatim,
@@ -87,7 +90,7 @@ def test_rescanning_an_unchanged_file_does_not_write_the_message_again(tmp_path)
     """A rescan is idempotent because the identity is derived from the content.
 
     `EvidenceInput.idempotency_key` is a hash of the source, conversation,
-    author, ordinal and content hash, and `evidence_id` is a uuid5 over that
+    protocol role and content hash, and `evidence_id` is a uuid5 over that
     key, so replaying an unchanged file reproduces the same primary key and
     `ON CONFLICT` absorbs it. Verified: removing the anchoring logic still
     leaves this green, so this pins the contract, not one implementation.
@@ -102,6 +105,87 @@ def test_rescanning_an_unchanged_file_does_not_write_the_message_again(tmp_path)
     assert store.count() == 1
 
 
+def test_derived_identity_uses_role_not_host_author_guess(tmp_path) -> None:
+    store = EvidenceStore(tmp_path / "one-memory.sqlite3")
+    first = store.put(evidence(message_id=None, written_by=WrittenBy.UNKNOWN))
+    replay = store.put(evidence(message_id=None, written_by=WrittenBy.PERSON))
+    assistant = store.put(evidence(message_id=None, role="assistant", written_by=WrittenBy.AGENT))
+
+    assert replay.record.evidence_id == first.record.evidence_id
+    assert replay.was_created is False
+    assert assistant.record.evidence_id != first.record.evidence_id
+    assert store.count() == 2
+
+
+def test_task_proof_ignores_legacy_written_by_metadata(tmp_path) -> None:
+    store = EvidenceStore(tmp_path / "one-memory.sqlite3")
+    record = store.put(evidence()).record
+    proof = evidence_ref(record)
+
+    changed_author = record.model_copy(update={"written_by": WrittenBy.AGENT})
+    legacy_identity = record.model_dump(
+        mode="json",
+        exclude={"origin_meta", "imported_at", "updated_at"},
+    )
+    legacy_proof = {
+        **proof,
+        "identity_hash": sha256(
+            json.dumps(legacy_identity, sort_keys=True).encode()
+        ).hexdigest(),
+    }
+
+    assert matches_evidence_ref(changed_author, proof) is True
+    assert matches_evidence_ref(changed_author, legacy_proof) is True
+
+
+def test_rescan_reuses_a_row_stored_under_written_by_derived_v3(tmp_path) -> None:
+    database = tmp_path / "one-memory.sqlite3"
+    store = EvidenceStore(database)
+    item = evidence(message_id=None, written_by=WrittenBy.UNKNOWN)
+    original = store.put(item).record
+    with sqlite3.connect(database) as connection, connection:
+        connection.execute(
+            "UPDATE evidence SET idempotency_key=? WHERE evidence_id=?",
+            (item.written_by_derived_v3_idempotency_key, original.evidence_id),
+        )
+
+    replay = store.put(evidence(message_id=None, written_by=WrittenBy.PERSON))
+
+    assert replay.was_created is False
+    assert replay.record.evidence_id == original.evidence_id
+    assert store.count() == 1
+
+
+def test_legacy_identity_role_change_revises_the_existing_evidence(tmp_path) -> None:
+    database = tmp_path / "one-memory.sqlite3"
+    store = EvidenceStore(database)
+    old = evidence(
+        message_id=None,
+        role="assistant",
+        written_by=WrittenBy.AGENT,
+    )
+    original = store.put(old).record
+    with sqlite3.connect(database) as connection, connection:
+        connection.execute(
+            "UPDATE evidence SET idempotency_key=? WHERE evidence_id=?",
+            (old.written_by_derived_v3_idempotency_key, original.evidence_id),
+        )
+
+    replay = store.put(
+        evidence(
+            message_id=None,
+            role="user",
+            written_by=WrittenBy.AGENT,
+        )
+    )
+
+    assert replay.record.evidence_id == original.evidence_id
+    assert replay.was_revised is True
+    assert replay.record.revision == original.revision + 1
+    assert replay.record.role == "user"
+    assert store.count() == 1
+
+
 def test_changed_text_at_one_position_is_a_new_message_not_a_silent_edit(tmp_path) -> None:
     """An edited line becomes new evidence, never an overwrite of the old one.
 
@@ -146,6 +230,7 @@ def test_personal_candidate_prefilter_excludes_unrelated_user_turns(tmp_path) ->
         evidence(
             message_id="assistant",
             verbatim="我偏好简洁文案。",
+            role="assistant",
             written_by=WrittenBy.AGENT,
         )
     )

memhub/apps/core/tests/test_hy_source_context.py

@@ -248,18 +248,8 @@ def test_profile_rendering_keeps_per_field_evidence_without_dropping_native_prof
     assert bindings.profile_support("The user's name is 陈.", {"name": "陈"}) == ()
 
 
-def test_bound_context_refuses_a_turn_with_a_wrong_author_pair(tmp_path):
-    """The author must match the role: only user/person, assistant/agent, tool/tool.
-
-    The objective requires 作者 to be truthful. The other scope tests only ever
-    build well-formed pairs, so this pins the pair check itself: an assistant
-    turn attributed to a person, or a tool turn attributed to one, must be
-    refused with `context_scope` rather than silently trusted.
-
-    `user` + AGENT is deliberately accepted since D98: the host parser marks a
-    sub agent's question that way, and that turn is sent like any other. What
-    is refused is the pair that cannot occur, not the one we now allow.
-    """
+def test_bound_context_uses_protocol_role_not_author_guess(tmp_path):
+    """Host author guesses never reject a valid user/assistant/tool message."""
     from one_memory.hy_source_context import HySourceContext, HyWriteBindings
 
     store = EvidenceStore(tmp_path / "evidence.db")
@@ -268,8 +258,7 @@ def test_bound_context_refuses_a_turn_with_a_wrong_author_pair(tmp_path):
 
     assistant = source(store, "reply", "assistant", "好的，继续推进。")
     wrong_author = assistant.model_copy(update={"written_by": WrittenBy.PERSON})
-    with pytest.raises(ValueError, match="context_scope"):
-        HyWriteBindings(HySourceContext(current=(wrong_author,)))
+    HyWriteBindings(HySourceContext(current=(wrong_author,)))
 
     # `source()` maps non-user roles to AGENT, so a tool turn needs its own build.
     tool = store.put(
@@ -284,7 +273,10 @@ def test_bound_context_refuses_a_turn_with_a_wrong_author_pair(tmp_path):
             origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
         )
     ).record
-    assert HyWriteBindings(HySourceContext(current=(tool,)))  # valid tool/tool pair
+    assert HyWriteBindings(HySourceContext(current=(tool,)))
     wrong_tool = tool.model_copy(update={"written_by": WrittenBy.PERSON})
+    HyWriteBindings(HySourceContext(current=(wrong_tool,)))
+
+    unsupported = assistant.model_copy(update={"role": "system"})
     with pytest.raises(ValueError, match="context_scope"):
-        HyWriteBindings(HySourceContext(current=(wrong_tool,)))
+        HyWriteBindings(HySourceContext(current=(unsupported,)))

memhub/apps/core/tests/test_ingest_journal.py

@@ -111,7 +111,7 @@ def test_import_metadata_changes_invalidate_without_authorizing_content(setup, c
 @pytest.mark.parametrize("cause", ["import", "maintenance"])
 @pytest.mark.parametrize(
     "column,value",
-    [("verbatim", "changed raw body"), ("written_by", "agent"), ("role", "assistant")],
+    [("verbatim", "changed raw body"), ("role", "assistant")],
 )
 def test_material_change_eligibility_depends_on_trusted_cause(setup, cause, column, value):
     db, store, _ = setup
@@ -126,6 +126,17 @@ def test_material_change_eligibility_depends_on_trusted_cause(setup, cause, colu
     assert new.kind == ("revise" if cause == "import" else "metadata")
 
 
+def test_legacy_written_by_change_is_not_a_material_event(setup):
+    db, store, _ = setup
+    db.execute("BEGIN IMMEDIATE")
+    key = put(store, db)
+    journal.initialize(db)
+    old = journal.current(db, key)
+    db.execute("UPDATE evidence SET written_by='agent' WHERE evidence_id=?", (key,))
+
+    assert journal.record(db, key, cause="import") == old
+
+
 def test_complete_origin_metadata_changes_and_observation_noise(setup):
     db, store, _ = setup
     db.execute("BEGIN IMMEDIATE")

memhub/apps/core/tests/test_memory_processing.py

@@ -1780,7 +1780,7 @@ def test_process_records_bounds_a_hung_model_call_by_the_profile_timeout(
     with sqlite3.connect(tmp_path / "one-memory.sqlite3") as db:
         db.executescript(
             """
-            CREATE TABLE source_grants (
+            CREATE TABLE IF NOT EXISTS source_grants (
                 user_id TEXT NOT NULL,
                 source_id TEXT NOT NULL,
                 root_path TEXT NOT NULL,
@@ -2038,6 +2038,126 @@ def test_oversized_single_turn_segments_are_each_processed(tmp_path) -> None:
     assert len(governance.list(user_id="local")) == 2
 
 
+def test_direct_processing_uses_the_supplied_frozen_revision(tmp_path) -> None:
+    stores = _setup(tmp_path)
+    evidence, profiles, tasks, governance, secrets, _profile = stores
+    old_user = _put(
+        evidence,
+        message_id="rewound-user",
+        role="user",
+        written_by=WrittenBy.PERSON,
+        content="old branch",
+    )
+    reply = _put(
+        evidence,
+        message_id="rewound-reply",
+        role="assistant",
+        written_by=WrittenBy.AGENT,
+        content="old reply",
+    )
+    revised = _put(
+        evidence,
+        message_id="rewound-user",
+        role="user",
+        written_by=WrittenBy.PERSON,
+        content="new branch",
+    )
+    assert revised.evidence_id == old_user.evidence_id
+    assert revised.revision == old_user.revision + 1
+    with sqlite3.connect(tasks.database_path) as connection, connection:
+        connection.execute(
+            """
+            CREATE TABLE source_grants (
+                user_id TEXT NOT NULL,
+                source_id TEXT NOT NULL,
+                root_path TEXT NOT NULL,
+                requested_path TEXT NOT NULL,
+                root_identity TEXT NOT NULL,
+                candidate_kind TEXT NOT NULL,
+                granted_at TEXT NOT NULL,
+                revoked_at TEXT,
+                revision INTEGER NOT NULL,
+                PRIMARY KEY (user_id, source_id)
+            )
+            """
+        )
+        connection.execute(
+            """
+            INSERT INTO source_grants (
+                user_id,source_id,root_path,requested_path,root_identity,
+                candidate_kind,granted_at,revoked_at,revision
+            ) VALUES (?,?,?,?,?,?,?,NULL,1)
+            """,
+            (
+                "local",
+                "cursor",
+                "/authorized",
+                "/authorized",
+                "test-root",
+                "default",
+                datetime.now(UTC).isoformat(),
+            ),
+        )
+    bridge = FakeBridge(outputs=())
+    supervisor = MemoryProcessingSupervisor(
+        tasks=tasks,
+        profiles=profiles,
+        secrets=secrets,
+        evidence=evidence,
+        governance=governance,
+        data_dir=tmp_path,
+        runner_factory=lambda: _runner(tmp_path, stores, bridge),
+    )
+
+    completed = asyncio.run(supervisor.process_records((old_user, reply)))
+
+    assert len(completed) == 1
+    assert [turn.content for turn in bridge.requests[0].turns] == [
+        "old branch",
+        "old reply",
+    ]
+
+
+def test_frozen_current_still_revalidates_history_after_the_model_call(tmp_path) -> None:
+    from one_memory.hy_source_context import HySourceContext
+
+    stores = _setup(tmp_path)
+    evidence, _profiles, tasks, _governance, _secrets, _profile = stores
+    current = _put(
+        evidence,
+        message_id="frozen-current",
+        role="user",
+        written_by=WrittenBy.PERSON,
+        content="old current",
+    )
+    history = _put(
+        evidence,
+        message_id="frozen-history",
+        role="assistant",
+        written_by=WrittenBy.AGENT,
+        content="stable history",
+    )
+    _put(
+        evidence,
+        message_id="frozen-current",
+        role="user",
+        written_by=WrittenBy.PERSON,
+        content="new current",
+    )
+    runner = _runner(tmp_path, stores, FakeBridge(outputs=()))
+    context = HySourceContext(current=(current,), history=(history,))
+
+    runner._check_source_context(context, skip_current=True)
+    with sqlite3.connect(tasks.database_path) as connection, connection:
+        connection.execute(
+            "DELETE FROM evidence WHERE evidence_id=?",
+            (history.evidence_id,),
+        )
+
+    with pytest.raises(HyMemoryBridgeError, match="hy_memory_evidence_changed"):
+        runner._check_source_context(context, skip_current=True)
+
+
 def test_oversized_assistant_segments_keep_the_question_as_context(tmp_path) -> None:
     evidence = EvidenceStore(tmp_path / "one-memory.sqlite3")
     question = _put(

memhub/apps/core/tests/test_memory_server_contracts.py

@@ -125,8 +125,7 @@ def test_write_job_state_covers_unknown_and_terminal() -> None:
     } == {state for state in WriteJobState if not state.is_terminal}
 
 
-def test_canonical_message_requires_author_and_ordinal() -> None:
-    """A message without a known author or position cannot be ordered later."""
+def test_canonical_message_requires_a_nonnegative_ordinal() -> None:
     with pytest.raises(ValueError):
         CanonicalMessage(
             harness_id="cursor",
@@ -134,6 +133,5 @@ def test_canonical_message_requires_author_and_ordinal() -> None:
             message_id=None,
             ordinal=-1,
             role="user",
-            written_by="person",
             content="hello",
         )

memhub/apps/core/tests/test_memory_server_durable_queue.py

@@ -10,6 +10,7 @@ import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
 from datetime import UTC, datetime, timedelta
+from hashlib import sha256
 
 import pytest
 
@@ -38,11 +39,13 @@ def _request(
     evidence: tuple[str, ...] = ("e1",),
     key: str = "idem-1",
     source: str | None = None,
+    session: str | None = None,
     event_at: datetime | None = None,
 ) -> WriteRequest:
-    batch = BatchIdentity(internal_session_id="sess-1", first_round=1, last_round=5)
+    session_id = session or f"sess-{source or '1'}"
+    batch = BatchIdentity(internal_session_id=session_id, first_round=1, last_round=5)
     return WriteRequest(
-        internal_session_id="sess-1",
+        internal_session_id=session_id,
         batch_identity=batch,
         evidence_ids=list(evidence),
         idempotency_key=key,
@@ -94,13 +97,8 @@ def test_acceptance_is_written_to_the_business_database(tmp_path, monkeypatch) -
     ]
 
 
-def test_the_stored_payload_is_references_only(tmp_path) -> None:
-    """The job payload must point at Evidence, never carry its text.
-
-    A queue row is durable and much longer-lived than the API call that made
-    it, so storing verbatim content here would be a second copy of the user's
-    private text. The payload is the request's identifiers only.
-    """
+def test_an_isolated_legacy_queue_keeps_the_empty_snapshot_compatibility_shape(tmp_path) -> None:
+    """Queues without an Evidence authority exist only for legacy/unit compatibility."""
     import json
 
     path = tmp_path / "one-memory.sqlite3"
@@ -112,10 +110,12 @@ def test_the_stored_payload_is_references_only(tmp_path) -> None:
         "internal_session_id",
         "batch_identity",
         "evidence_ids",
+        "input_records",
         "idempotency_key",
         "priority",
         "source_id",
     }
+    assert payload["input_records"] == []
     assert payload["evidence_ids"] == ["ev-1", "ev-2"]
 
 
@@ -216,20 +216,17 @@ def test_a_replay_is_idempotent_even_after_its_evidence_is_removed(tmp_path) ->
     assert replay.enqueued is False
     assert len(_rows(path)) == 1
 
-
 def test_a_settled_job_replays_as_a_hit_not_a_conflict(tmp_path) -> None:
-    """Replaying a completed key must return the job, never raise.
-
-    Real trigger: 236 startup replays of one key, all
-    `ValueError: idempotency key ... already exists with different content`.
-    The stored payload had been reaped to a digest sentinel, and the caller's
-    key is derived from the evidence's revision, which can move on between
-    replays while the payload's evidence ids stay put. Refusing that pair
-    turned a normal startup replay into an exception every time.
-    """
+    """A historical reaped payload matches only its exact pre-snapshot intent."""
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
-    first = queue.submit(_request(evidence=("e1",), key="idem-settled"))
+    request = _request(evidence=("e1",), key="idem-settled")
+    first = queue.submit(request)
+    legacy = request.model_dump(
+        mode="json",
+        exclude={"event_at", "time_quality", "input_records"},
+    )
+    digest = sha256(json.dumps(legacy, sort_keys=True).encode()).hexdigest()
 
     # Settle it the way the drain does: terminal state, HY invoked, and the
     # payload reaped down to its digest so no body is kept.
@@ -238,19 +235,40 @@ def test_a_settled_job_replays_as_a_hit_not_a_conflict(tmp_path) -> None:
             "UPDATE memory_write_jobs SET state=?, hy_invoked=1, payload_json=? WHERE job_id=?",
             (
                 COMPLETED,
-                json.dumps({"_one_memory_reaped_payload": "0" * 64}),
+                json.dumps({"_one_memory_reaped_payload": digest}),
                 first.job_id,
             ),
         )
 
-    # A payload that cannot match the reaped digest, under the same key.
-    replay = queue.submit(_request(evidence=("e1",), key="idem-settled"))
+    replay = queue.submit(request)
 
     assert replay.job_id == first.job_id
     assert replay.state == COMPLETED
     assert replay.enqueued is False
     assert len(_rows(path)) == 1
 
+    with pytest.raises(ValueError, match="different content"):
+        queue.submit(
+            _request(
+                evidence=("e1",),
+                key="idem-settled",
+                source="cursor",
+                session="different-session",
+            )
+        )
+
+
+def test_a_real_completed_job_still_rejects_the_same_key_with_different_input(
+    tmp_path,
+) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    first = queue.submit(_request(evidence=("e1",), key="completed-conflict"))
+    queue.settle(first.job_id, state=COMPLETED, hy_invoked=True)
+
+    with pytest.raises(ValueError, match="different content"):
+        queue.submit(_request(evidence=("e2",), key="completed-conflict"))
+
 
 def test_a_killed_worker_leaves_a_possibly_sent_job_unknown_after_restart(tmp_path) -> None:
     """A real process death after HY was invoked must not become retryable.
@@ -1144,10 +1162,13 @@ def test_unresolved_unknown_is_automatically_retried_only_once(tmp_path) -> None
     first_retry = queue.get(record.request_id)
     assert first_retry.state == PENDING
     with sqlite3.connect(path) as connection:
-        assert connection.execute(
-            "SELECT attempts FROM memory_write_jobs WHERE job_id=?",
-            (record.job_id,),
-        ).fetchone()[0] == 1
+        assert (
+            connection.execute(
+                "SELECT attempts FROM memory_write_jobs WHERE job_id=?",
+                (record.job_id,),
+            ).fetchone()[0]
+            == 1
+        )
 
     drain = DurableWriteDrain(queue, execute=_never_execute, owner="retry-owner")
     assert drain.claim(limit=1) == (record.job_id,)
@@ -1161,10 +1182,13 @@ def test_unresolved_unknown_is_automatically_retried_only_once(tmp_path) -> None
     )
 
     with sqlite3.connect(path) as connection:
-        assert connection.execute(
-            "SELECT attempts FROM memory_write_jobs WHERE job_id=?",
-            (record.job_id,),
-        ).fetchone()[0] == 2
+        assert (
+            connection.execute(
+                "SELECT attempts FROM memory_write_jobs WHERE job_id=?",
+                (record.job_id,),
+            ).fetchone()[0]
+            == 2
+        )
     assert (
         queue.retry_first_unknown_after_reconcile(
             older_than=datetime.now(UTC) + timedelta(minutes=1)
@@ -2541,21 +2565,16 @@ def test_settle_with_an_owner_does_not_touch_an_ownerless_row(tmp_path, caplog)
     assert not any("durable_write_settle_missed_row" in message for message in messages), messages
 
 
-def test_terminal_payloads_are_reaped_but_replay_still_detects_conflicts(tmp_path) -> None:
-    """Payload cleanup must not weaken idempotency.
-
-    Once a job is terminal the serialized request is dead weight, so it is
-    replaced by a digest. Replaying the same intent must still resolve to the
-    same row, and the same key with different content must still be refused.
-    """
+def test_terminal_payloads_are_retained_and_replay_still_detects_conflicts(tmp_path) -> None:
+    """Terminal input is immutable audit data and idempotency remains strict."""
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     record = queue.submit(_request(evidence=("e1",)))
     queue.settle(record.job_id, state="completed")
 
-    assert queue.clear_terminal_payloads() == 1
+    assert queue.clear_terminal_payloads() == 0
     stored = _rows(path)[0]["payload_json"]
-    assert "e1" not in stored and "_one_memory_reaped_payload" in stored
+    assert "e1" in stored and "_one_memory_reaped_payload" not in stored
     assert _rows(path)[0]["evidence_ids_json"] == '["e1"]'
 
     # Same intent still resolves to the existing row, not a second write.
@@ -2569,15 +2588,15 @@ def test_terminal_payloads_are_reaped_but_replay_still_detects_conflicts(tmp_pat
         queue.submit(_request(evidence=("e2",)))
 
 
-def test_terminal_payload_reaping_is_bounded_and_idempotent(tmp_path) -> None:
+def test_terminal_payload_cleanup_is_a_compatibility_noop(tmp_path) -> None:
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     for index in range(3):
         record = queue.submit(_request(evidence=("e1",), key=f"idem-{index}"))
         queue.settle(record.job_id, state="dead")
 
-    assert queue.clear_terminal_payloads(limit=2) == 2
-    assert queue.clear_terminal_payloads(limit=2) == 1
+    assert queue.clear_terminal_payloads(limit=2) == 0
+    assert queue.clear_terminal_payloads(limit=2) == 0
     assert queue.clear_terminal_payloads(limit=2) == 0
 
 
@@ -2596,8 +2615,8 @@ def test_unknown_and_pending_payloads_are_never_reaped(tmp_path) -> None:
     assert "e2" in payloads[unknown.request_id]
 
 
-def test_reaped_terminal_payload_does_not_reappear_as_an_unsubmitted_orphan(tmp_path) -> None:
-    """Payload cleanup must not make a completed Evidence look unqueued.
+def test_retained_terminal_payload_does_not_reappear_as_an_unsubmitted_orphan(tmp_path) -> None:
+    """A completed Evidence remains queued while its full input is retained.
 
     The reconcile primitive scans the durable queue by Evidence id. If it
     derived that set from `payload_json`, reaping a terminal row would erase
@@ -2631,7 +2650,7 @@ def test_reaped_terminal_payload_does_not_reappear_as_an_unsubmitted_orphan(tmp_
     queue = DurableWriteQueue(path)
     record = queue.submit(_request(evidence=(evidence.evidence_id,), key="idem-reaped"))
     queue.settle(record.job_id, state="completed")
-    assert queue.clear_terminal_payloads() == 1
+    assert queue.clear_terminal_payloads() == 0
     assert queue.unsubmitted_evidence_ids() == ()
 
     orphan = store.put(
@@ -2649,19 +2668,17 @@ def test_reaped_terminal_payload_does_not_reappear_as_an_unsubmitted_orphan(tmp_
     assert queue.unsubmitted_evidence_ids() == (orphan.evidence_id,)
 
 
-def test_payload_for_a_reaped_terminal_row_returns_none(tmp_path) -> None:
-    """A stale claim of a reaped row must not resurrect a dead intent."""
+def test_payload_for_a_terminal_row_keeps_the_immutable_input(tmp_path) -> None:
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     record = queue.submit(_request())
     queue.settle(record.job_id, state="completed")
     queue.clear_terminal_payloads()
 
-    assert queue.payload_for(record.job_id) is None
+    assert queue.payload_for(record.job_id) == _request()
 
 
-def test_drain_reaps_terminal_payloads_when_idle(tmp_path) -> None:
-    """The reaper must actually run in the worker, not only in tests."""
+def test_drain_retains_terminal_payloads_when_idle(tmp_path) -> None:
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     record = queue.submit(_request())
@@ -2686,19 +2703,11 @@ def test_drain_reaps_terminal_payloads_when_idle(tmp_path) -> None:
     asyncio.run(scenario())
 
     assert executed == []
-    assert "_one_memory_reaped_payload" in _rows(path)[0]["payload_json"]
+    assert "_one_memory_reaped_payload" not in _rows(path)[0]["payload_json"]
 
 
-def test_the_drain_reaps_terminal_payloads_even_while_never_idle(tmp_path) -> None:
-    """Terminal payloads must be reaped even if the queue never goes idle.
-
-    Regression: `clear_terminal_payloads` ran only on the `executed == 0` idle
-    branch, so a sustained backlog kept every finished row's full serialized
-    request body (which holds the written evidence references) in the business
-    database indefinitely — a disk/privacy-retention invariant starved exactly
-    when the queue is busiest. Reaping is a bounded maintenance write and must
-    be gated by its own interval, not by idleness.
-    """
+def test_the_drain_retains_terminal_payloads_while_never_idle(tmp_path) -> None:
+    """A sustained backlog must not mutate an earlier job's input snapshot."""
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     finished = queue.submit(_request(key="finished"))
@@ -2726,9 +2735,8 @@ def test_the_drain_reaps_terminal_payloads_even_while_never_idle(tmp_path) -> No
 
     payload = _rows(path)
     by_key = {row["idempotency_key"]: row["payload_json"] for row in payload}
-    assert "_one_memory_reaped_payload" in by_key["finished"], (
-        "a completed row's payload must be reaped even under a standing backlog"
-    )
+    assert "_one_memory_reaped_payload" not in by_key["finished"]
+    assert '"e1"' in by_key["finished"]
 
 
 def test_the_drain_readmits_model_waiters_even_while_never_idle(tmp_path) -> None:
@@ -3177,8 +3185,8 @@ def test_claim_budget_ab_against_a_held_writer(tmp_path) -> None:
 def test_idle_maintenance_does_not_write_every_poll(tmp_path, monkeypatch) -> None:
     """An empty queue must not issue a write per idle poll.
 
-    `clear_terminal_payloads` and `readmit_model_waiters` both write. On an
-    empty queue the loop wakes every `idle_pause`; running them each time would
+    `readmit_model_waiters` writes. On an empty queue the loop wakes every
+    `idle_pause`; running it each time would
     contend with imports for the single writer for no benefit. They share a
     coarse interval gate, so a 20-poll window performs at most a couple of
     writes.
@@ -3187,22 +3195,14 @@ def test_idle_maintenance_does_not_write_every_poll(tmp_path, monkeypatch) -> No
     queue = DurableWriteQueue(path)
     drain = DurableWriteDrain(queue, execute=_never_execute)
 
-    reaps = 0
     readmits = 0
-    real_reap = queue.clear_terminal_payloads
     real_readmit = queue.readmit_model_waiters
 
-    def counted_reap(*args, **kwargs):
-        nonlocal reaps
-        reaps += 1
-        return real_reap(*args, **kwargs)
-
     def counted_readmit(*args, **kwargs):
         nonlocal readmits
         readmits += 1
         return real_readmit(*args, **kwargs)
 
-    monkeypatch.setattr(queue, "clear_terminal_payloads", counted_reap)
     monkeypatch.setattr(queue, "readmit_model_waiters", counted_readmit)
     monkeypatch.setattr(drain, "claim", lambda **_kw: ())
 
@@ -3216,7 +3216,6 @@ def test_idle_maintenance_does_not_write_every_poll(tmp_path, monkeypatch) -> No
 
     # Well over 20 polls happened; maintenance ran at most once (the interval
     # gate), not once per poll.
-    assert reaps <= 1
     assert readmits <= 1
 
 
@@ -3347,7 +3346,7 @@ def test_idle_maintenance_database_failure_is_sampled_without_database_error(
     def locked(*_args, **_kwargs):
         raise sqlite3.OperationalError("database is locked")
 
-    monkeypatch.setattr(drain.queue, "clear_terminal_payloads", locked)
+    monkeypatch.setattr(drain.queue, "readmit_model_waiters", locked)
     monkeypatch.setattr(drain, "claim", lambda **_kw: ())
 
     async def scenario() -> None:
@@ -3363,8 +3362,8 @@ def test_idle_maintenance_database_failure_is_sampled_without_database_error(
     def programming_error(*_args, **_kwargs):
         raise RuntimeError("maintenance bug")
 
-    monkeypatch.setattr(drain.queue, "clear_terminal_payloads", programming_error)
-    drain._last_terminal_reap = None  # noqa: SLF001 - force the next maintenance tick
+    monkeypatch.setattr(drain.queue, "readmit_model_waiters", programming_error)
+    drain._last_model_readmit = None  # noqa: SLF001 - force the next maintenance tick
 
     asyncio.run(scenario())
 
@@ -4288,9 +4287,7 @@ def test_event_time_window_claims_cross_harness_jobs_within_ten_minutes(
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     anchor = datetime(2026, 10, 18, 8, 0, tzinfo=UTC)
-    first = queue.submit(
-        _request(key="window-a", source="cursor", event_at=anchor)
-    )
+    first = queue.submit(_request(key="window-a", source="cursor", event_at=anchor))
     inside = queue.submit(
         _request(
             key="window-b",
@@ -4334,9 +4331,7 @@ def test_event_time_window_anchor_does_not_slide_while_a_job_is_running(
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     anchor = datetime(2026, 10, 18, 8, 0, tzinfo=UTC)
-    first = queue.submit(
-        _request(key="fixed-a", source="cursor", event_at=anchor)
-    )
+    first = queue.submit(_request(key="fixed-a", source="cursor", event_at=anchor))
     inside = queue.submit(
         _request(
             key="fixed-b",
@@ -4815,18 +4810,10 @@ def test_readmit_paths_never_requeue_a_possibly_sent_row(tmp_path) -> None:
     assert DurableWriteDrain(queue, execute=_never_execute).claim(limit=5) == ()
 
 
-def test_a_refused_source_context_fails_instead_of_parking_as_unknown(
+def test_a_refused_source_context_is_nonblocking_unknown(
     tmp_path,
 ) -> None:
-    """A refusal raised before the call must not read as a possibly-sent write.
-
-    Issue #44: an OpenCode import carried `role=user, written_by=unknown`, so
-    `HyWriteBindings` — built inside `ingest`, ahead of the payload reaching HY
-    — refused the batch. The invocation marker is recorded before that call, so
-    the row was parked as `unknown`: a state that says "HY may hold a write we
-    cannot confirm" about a request that was never sent. It kept the reason off
-    the row and stopped the queue behind it.
-    """
+    """An ambiguous call-boundary failure is retained but never blocks FIFO."""
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     record = queue.submit(_request(key="refused-scope", source="opencode"))
@@ -4848,14 +4835,12 @@ def test_a_refused_source_context_fails_instead_of_parking_as_unknown(
 
     with sqlite3.connect(path.as_uri() + "?mode=ro", uri=True) as db:
         state, invoked, last_error = db.execute(
-            "SELECT state, hy_invoked, last_error FROM memory_write_jobs "
-            "WHERE job_id=?",
+            "SELECT state, hy_invoked, last_error FROM memory_write_jobs WHERE job_id=?",
             (record.job_id,),
         ).fetchone()
 
-    assert state == FAILED
-    # Nothing was sent, so reclaim must not treat this as possibly-sent.
-    assert invoked == 0
+    assert state == UNKNOWN
+    assert invoked == 1
     assert last_error == "hy_memory_write_verification_context_scope"
     assert DurableWriteDrain(queue, execute=_never_execute).claim(limit=1) == ()
 
@@ -4873,9 +4858,7 @@ def test_a_failure_after_the_call_still_parks_as_unknown(tmp_path) -> None:
             record.job_id,
             queue=queue,
             supervisor=_Recorder(
-                error=HyMemoryBridgeError(
-                    "hy_memory_write_verification_authority_invalid"
-                ),
+                error=HyMemoryBridgeError("hy_memory_write_verification_authority_invalid"),
                 invoke_hy=True,
             ),
             evidence=_EvidenceStore({"e1"}),
@@ -4893,40 +4876,27 @@ def test_a_failure_after_the_call_still_parks_as_unknown(tmp_path) -> None:
     assert invoked == 1
 
 
-def test_a_window_holding_only_blocked_rows_advances_instead_of_starving(
+def test_event_time_not_enqueue_order_drives_the_window(
     tmp_path,
 ) -> None:
-    """A window whose remaining rows FIFO holds back must be retired.
-
-    Live shape: the anchor sat on 2026-10-07T03:24Z while the four pending rows
-    inside it were each blocked by 900+ earlier siblings of the same Harness
-    whose own event times sat outside the range (the earliest was 2026-03-12).
-    None of them could ever be claimed from inside that window, and because the
-    window only asked whether the range still held *some* pending row it never
-    retired: 3.8k queued turns sat at zero completions until restart, which
-    does not help because the anchor is persisted.
-    """
+    """A later-enqueued old turn runs before an earlier-enqueued new turn."""
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
     anchor = datetime(2026, 10, 18, 8, 0, tzinfo=UTC)
 
-    # The anchor itself: another Harness, claimable, inside the range.
-    opener = queue.submit(
-        _request(key="starve-opener", source="codex", event_at=anchor)
-    )
-    # Same Harness as the blocked row but an hour later: outside the range, and
-    # it sorts earlier in queue_seq, so FIFO holds the row below behind it.
-    older_seq_newer_time = queue.submit(
+    enqueued_first = queue.submit(
         _request(
-            key="starve-head",
+            key="newer-event",
             source="workbuddy",
+            session="newer-session",
             event_at=anchor + timedelta(hours=1),
         )
     )
-    blocked = queue.submit(
+    enqueued_second = queue.submit(
         _request(
-            key="starve-blocked",
+            key="older-event",
             source="workbuddy",
+            session="older-session",
             event_at=anchor + timedelta(minutes=5),
         )
     )
@@ -4938,36 +4908,70 @@ def test_a_window_holding_only_blocked_rows_advances_instead_of_starving(
         max_concurrent_sources=10,
     )
 
-    # The blocked row is inside the window but FIFO-refused, so it is not
-    # claimed alongside the opener.
-    assert drain.claim(limit=10) == (opener.job_id,)
+    assert drain.claim(limit=10) == (enqueued_second.job_id,)
     queue.settle(
-        opener.job_id,
+        enqueued_second.job_id,
         state=COMPLETED,
         hy_invoked=True,
         owner="starve-owner",
     )
-
-    # The window now holds exactly one row and that row can never be claimed
-    # from inside it. Retire the window and re-anchor on the blocker.
-    assert drain.claim(limit=10) == (older_seq_newer_time.job_id,)
+    assert drain.claim(limit=10) == (enqueued_first.job_id,)
 
     with sqlite3.connect(path.as_uri() + "?mode=ro", uri=True) as db:
         window = db.execute(
-            "SELECT window_anchor FROM memory_write_schedule_state "
-            "WHERE singleton=1"
+            "SELECT window_anchor FROM memory_write_schedule_state WHERE singleton=1"
         ).fetchone()
     assert window is not None
-    assert str(window[0]) == (
-        anchor + timedelta(hours=1)
-    ).astimezone(UTC).isoformat()
+    assert str(window[0]) == (anchor + timedelta(hours=1)).astimezone(UTC).isoformat()
 
+
+def test_same_session_uses_event_time_then_queue_sequence(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    base = datetime(2026, 10, 18, 8, 0, tzinfo=UTC)
+    later = queue.submit(
+        _request(key="same-session-later", source="cursor", event_at=base + timedelta(hours=1))
+    )
+    earlier = queue.submit(_request(key="same-session-earlier", source="cursor", event_at=base))
+    drain = DurableWriteDrain(
+        queue,
+        execute=_never_execute,
+        owner="event-owner",
+        max_concurrent_sources=10,
+    )
+
+    assert drain.claim(limit=10) == (earlier.job_id,)
     queue.settle(
-        older_seq_newer_time.job_id,
+        earlier.job_id,
         state=COMPLETED,
         hy_invoked=True,
-        owner="starve-owner",
+        owner="event-owner",
     )
-    # Once the blocker is gone the deferred row keeps its original event time
-    # and is reached on the next pass.
-    assert drain.claim(limit=10) == (blocked.job_id,)
+    assert drain.claim(limit=10) == (later.job_id,)
+
+
+def test_equal_event_time_uses_queue_sequence_inside_one_session(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    event_at = datetime(2026, 10, 18, 8, 0, tzinfo=UTC)
+    first = queue.submit(
+        _request(key="equal-first", source="cursor", event_at=event_at)
+    )
+    second = queue.submit(
+        _request(key="equal-second", source="cursor", event_at=event_at)
+    )
+    drain = DurableWriteDrain(
+        queue,
+        execute=_never_execute,
+        owner="equal-owner",
+        max_concurrent_sources=10,
+    )
+
+    assert drain.claim(limit=10) == (first.job_id,)
+    queue.settle(
+        first.job_id,
+        state=COMPLETED,
+        hy_invoked=True,
+        owner="equal-owner",
+    )
+    assert drain.claim(limit=10) == (second.job_id,)

memhub/apps/core/tests/test_memory_server_migrations.py

@@ -2103,7 +2103,7 @@ def test_migration_49_resets_only_cursor_history_when_conversation_time_is_missi
 
     result = asyncio.run(migrate(SqliteMigrationStore(path), MIGRATIONS))
 
-    assert result.applied == (49, 50, 51, 52, 53)
+    assert result.applied == (49, 50, 51, 52, 53, 54)
     with main_database(path).read() as connection:
         assert connection.execute("SELECT COUNT(*) FROM source_cursors").fetchone()[0] == 0
         assert connection.execute("SELECT COUNT(*) FROM live_source_cursors").fetchone()[0] == 1
@@ -2136,7 +2136,7 @@ def test_upgrade_drops_obsolete_hy_node_permissions_without_touching_memories(
 
     result = asyncio.run(migrate(SqliteMigrationStore(path), MIGRATIONS))
 
-    assert result.applied == (51, 52, 53)
+    assert result.applied == (51, 52, 53, 54)
     with sqlite3.connect(path) as connection:
         tables = {
             row[0]
@@ -2217,3 +2217,55 @@ def test_migration_53_backfills_write_job_event_time_from_person_turn(
             "SELECT 1 FROM sqlite_master "
             "WHERE type='table' AND name='memory_write_schedule_state'"
         ).fetchone() == (1,)
+
+
+def test_migration_54_preserves_legacy_possibly_sent_tasks(tmp_path) -> None:
+    import sqlite3
+
+    from one_memory.memory_server.migration_runner import SqliteMigrationStore
+
+    path = tmp_path / "one-memory.sqlite3"
+    through_53 = tuple(item for item in MIGRATIONS if item.version <= 53)
+    asyncio.run(migrate(SqliteMigrationStore(path), through_53))
+    with sqlite3.connect(path) as connection, connection:
+        connection.execute(
+            """
+            INSERT INTO memory_tasks (
+                task_id,request_id,user_id,kind,input_ref,status,
+                profile_id,profile_revision,profile_snapshot_json,
+                attempts,created_at,updated_at,hy_invocation_count
+            ) VALUES (?,?,?,?,?,?,?,?,?,0,?,?,0)
+            """,
+            (
+                "legacy-sent",
+                "legacy-sent-request",
+                "local",
+                "extract",
+                "evidence",
+                "running",
+                "profile",
+                1,
+                "{}",
+                "2026-10-18T00:00:00+00:00",
+                "2026-10-18T00:00:00+00:00",
+            ),
+        )
+        connection.execute(
+            "CREATE TABLE memory_task_continuous_permissions "
+            "(task_id TEXT PRIMARY KEY)"
+        )
+        connection.execute(
+            "INSERT INTO memory_task_continuous_permissions VALUES (?)",
+            ("legacy-sent",),
+        )
+
+    asyncio.run(migrate(SqliteMigrationStore(path), MIGRATIONS))
+
+    with sqlite3.connect(path) as connection:
+        assert connection.execute(
+            "SELECT hy_invocation_count FROM memory_tasks WHERE task_id='legacy-sent'"
+        ).fetchone() == (1,)
+        assert connection.execute(
+            "SELECT 1 FROM sqlite_master "
+            "WHERE type='table' AND name='memory_task_continuous_permissions'"
+        ).fetchone() is None

memhub/apps/core/tests/test_memory_server_unified_entry.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import json
 import sqlite3
 from datetime import UTC, datetime
 
@@ -9,13 +11,23 @@ import pytest
 
 from one_memory.imports.models import EvidenceInput, EvidenceOrigin, OriginMetadata, WrittenBy
 from one_memory.imports.storage import EvidenceStore
-from one_memory.memory_server.durable_queue import DurableWriteQueue
+from one_memory.memory_server.durable_queue import (
+    DurableWriteDrain,
+    DurableWriteQueue,
+    execute_job,
+)
 from one_memory.memory_server.unified_entry import (
     UnifiedWriteIntake,
 )
 
 
-def _record(store: EvidenceStore, *, conversation_id: str = "c1", message_id: str = "m1"):
+def _record(
+    store: EvidenceStore,
+    *,
+    conversation_id: str = "c1",
+    message_id: str = "m1",
+    verbatim: str = "durable intake",
+):
     return store.put(
         EvidenceInput(
             source_id="cursor",
@@ -24,7 +36,7 @@ def _record(store: EvidenceStore, *, conversation_id: str = "c1", message_id: st
             role="user",
             written_by=WrittenBy.PERSON,
             created_at=datetime.now(UTC),
-            verbatim="durable intake",
+            verbatim=verbatim,
             origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
         )
     ).record
@@ -172,6 +184,136 @@ def test_replaying_the_same_evidence_does_not_create_a_second_row(tmp_path) -> N
     assert replay.enqueued is False
 
 
+def test_rewound_harness_revision_appends_a_job_with_immutable_input(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    store = EvidenceStore(path)
+    intake = UnifiedWriteIntake(DurableWriteQueue(path, evidence=store))
+    old = _record(store, verbatim="old branch")
+    old_job = intake.submit_evidence(old, first_round=1, last_round=1)
+
+    revised = _record(store, verbatim="new branch")
+    new_job = intake.submit_evidence(revised, first_round=1, last_round=1)
+
+    assert revised.evidence_id == old.evidence_id
+    assert revised.revision == old.revision + 1
+    assert new_job.record.job_id != old_job.record.job_id
+    assert intake.queue.payload_for(old_job.record.job_id).input_records[0].verbatim == "old branch"
+    assert intake.queue.payload_for(new_job.record.job_id).input_records[0].verbatim == "new branch"
+    assert intake.queue.unsubmitted_evidence_ids() == ()
+    with sqlite3.connect(path) as connection:
+        assert connection.execute("SELECT COUNT(*) FROM memory_write_jobs").fetchone()[0] == 2
+
+
+def test_revision_committed_before_job_crash_is_discovered_for_reconcile(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    store = EvidenceStore(path)
+    queue = DurableWriteQueue(path, evidence=store)
+    intake = UnifiedWriteIntake(queue)
+    old = _record(store, verbatim="old branch")
+    intake.submit_evidence(old, first_round=1, last_round=1)
+
+    revised = _record(store, verbatim="new branch")
+
+    assert revised.evidence_id == old.evidence_id
+    assert revised.revision == old.revision + 1
+    assert queue.unsubmitted_evidence_ids() == (revised.evidence_id,)
+
+
+def test_late_old_revision_snapshot_does_not_cover_the_current_revision(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    store = EvidenceStore(path)
+    queue = DurableWriteQueue(path, evidence=store)
+    intake = UnifiedWriteIntake(queue)
+    old = _record(store, verbatim="old branch")
+    revised = _record(store, verbatim="new branch")
+
+    intake.submit_evidence(old, first_round=1, last_round=1)
+
+    assert revised.evidence_id == old.evidence_id
+    assert revised.revision == old.revision + 1
+    assert queue.unsubmitted_evidence_ids() == (revised.evidence_id,)
+
+
+def test_old_branch_job_executes_its_frozen_revision_after_harness_rewind(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    store = EvidenceStore(path)
+    queue = DurableWriteQueue(path, evidence=store)
+    intake = UnifiedWriteIntake(queue)
+    old = _record(store, verbatim="old branch")
+    old_job = intake.submit_evidence(old, first_round=1, last_round=1)
+    revised = _record(store, verbatim="new branch")
+    intake.submit_evidence(revised, first_round=1, last_round=1)
+    drain = DurableWriteDrain(queue, execute=lambda _job_id: asyncio.sleep(0), owner="frozen")
+    assert drain.claim(limit=1) == (old_job.record.job_id,)
+
+    captured = []
+
+    class Supervisor:
+        async def process_records(self, records, *, before_hy_invocation):
+            captured.extend(records)
+            await before_hy_invocation()
+
+    asyncio.run(
+        execute_job(
+            old_job.record.job_id,
+            queue=queue,
+            supervisor=Supervisor(),
+            evidence=store,
+            owner="frozen",
+        )
+    )
+
+    assert [record.verbatim for record in captured] == ["old branch"]
+    assert queue.get(old_job.request_id).state == "completed"
+
+
+def test_reopening_reads_an_old_id_only_job_without_rewriting_it(tmp_path) -> None:
+    from one_memory.memory_server.contracts import BatchIdentity, WriteRequest
+
+    path = tmp_path / "one-memory.sqlite3"
+    store = EvidenceStore(path)
+    record = _record(store)
+    queue = DurableWriteQueue(path, evidence=store)
+    legacy = queue.submit(
+        WriteRequest(
+            internal_session_id=record.conversation_id,
+            source_id=record.source_id,
+            batch_identity=BatchIdentity(
+                internal_session_id=record.conversation_id,
+                first_round=1,
+                last_round=1,
+            ),
+            evidence_ids=[record.evidence_id],
+            idempotency_key="legacy-id-only",
+        )
+    )
+    with sqlite3.connect(path) as connection, connection:
+        payload = json.loads(
+            connection.execute(
+                "SELECT payload_json FROM memory_write_jobs WHERE job_id=?",
+                (legacy.job_id,),
+            ).fetchone()[0]
+        )
+        payload.pop("input_records", None)
+        connection.execute(
+            "UPDATE memory_write_jobs SET payload_json=? WHERE job_id=?",
+            (json.dumps(payload, sort_keys=True), legacy.job_id),
+        )
+    assert queue.payload_for(legacy.job_id).input_records == ()
+
+    reopened = DurableWriteQueue(path, evidence=store)
+
+    assert reopened.payload_for(legacy.job_id).input_records == ()
+    with sqlite3.connect(path) as connection:
+        persisted = json.loads(
+            connection.execute(
+                "SELECT payload_json FROM memory_write_jobs WHERE job_id=?",
+                (legacy.job_id,),
+            ).fetchone()[0]
+        )
+    assert "input_records" not in persisted
+
+
 def test_committed_evidence_without_submit_is_the_red_state(tmp_path) -> None:
     """The exact crash window the C0 wiring has to close.
 

memhub/apps/core/tests/test_memory_tasks.py

@@ -1500,14 +1500,8 @@ def test_restart_splits_a_sent_continuous_task_from_an_unsent_one(tmp_path) -> N
         input_ref="evidence-unsent",
         profile=profile,
     )
-    with sqlite3.connect(store.database_path) as connection, connection:
-        connection.execute(
-            "CREATE TABLE memory_task_continuous_permissions (task_id TEXT PRIMARY KEY)"
-        )
-        connection.execute(
-            "INSERT INTO memory_task_continuous_permissions VALUES (?)", (sent.task_id,)
-        )
-    store.claim(sent.task_id)
+    _sent_task, sent_token = store.claim(sent.task_id)
+    store.record_hy_invocation(sent.task_id, runner_token=sent_token)
     store.claim(unsent.task_id)
 
     recovered = store.recover_running()

memhub/apps/core/tests/test_retrieval_mapped_content.py

@@ -132,19 +132,22 @@ def test_raw_authors_keep_exact_body_and_do_not_mutate_evidence(role, author):
 @pytest.mark.parametrize(
     "change",
     [
-        {"role": "user", "written_by": WrittenBy.AGENT},
-        {"role": "assistant", "written_by": WrittenBy.PERSON},
         {"role": "tool", "written_by": WrittenBy.TOOL},
         {"role": "system"},
-        {"written_by": WrittenBy.UNKNOWN},
         {"content_kind": EvidenceContentKind.MEMORY_CONTENT},
     ],
 )
-def test_raw_and_author_qualification_rejects_without_tokenizing(change):
+def test_raw_role_qualification_rejects_without_tokenizing(change):
     with pytest.raises(MappedContentError, match="mapped_content_ineligible"):
         prepare_mapped_content(record("ordinary body", **change))
 
 
+@pytest.mark.parametrize("author", list(WrittenBy))
+def test_host_author_guess_does_not_change_protocol_role_qualification(author):
+    content = prepare_mapped_content(record("ordinary body", role="user", written_by=author))
+    assert content.state == "ready"
+
+
 @pytest.mark.parametrize(
     "text,reason",
     [

memhub/apps/core/tests/test_scanner_sessionizer.py

@@ -25,7 +25,6 @@ def _message(
         message_id=message_id or f"m{ordinal}",
         ordinal=ordinal,
         role=role,
-        written_by="person" if role == "user" else "model",
         content=content,
     )
 

memhub/apps/core/tests/test_source_import_receipts.py

@@ -39,7 +39,11 @@ def record(
         source_id=source,
         conversation_id=f"c{index % 2}",
         message_id=str(index),
-        role="assistant" if author == WrittenBy.AGENT else "user",
+        role=(
+            "assistant"
+            if author == WrittenBy.AGENT
+            else ("tool" if author == WrittenBy.TOOL else "user")
+        ),
         written_by=author,
         created_at=time,
         verbatim=text or f"original {index}",
@@ -83,8 +87,8 @@ def test_receipt_counts_saved_records_without_inventing_dates_or_other_sources(t
             data = response.json()
             assert data["source_id"] == "claude-code"
             assert data["messages"] == 5 and data["conversations"] == 2
-            assert data["person_messages"] == 2 and data["agent_messages"] == 1
-            assert data["unknown_author_messages"] == 1 and data["tool_messages"] == 1
+            assert data["person_messages"] == 3 and data["agent_messages"] == 1
+            assert data["unknown_author_messages"] == 0 and data["tool_messages"] == 1
             assert data["unknown_time_messages"] == 4
             assert len(data["examples"]) == 3
             assert "excluded-source" not in response.text and "excluded-user" not in response.text

memhub/apps/core/tests/test_unanswered_turn.py

@@ -34,8 +34,8 @@ def test_a_heartbeat_is_not_a_question():
     assert last_row_opens_an_unanswered_turn(("user", "person", "<heartbeat>ping")) is False
 
 
-def test_a_message_written_by_the_agent_as_user_is_not_a_question():
-    assert last_row_opens_an_unanswered_turn(("user", "agent", "继续")) is False
+def test_protocol_user_role_opens_a_turn_regardless_of_legacy_author_guess():
+    assert last_row_opens_an_unanswered_turn(("user", "agent", "继续")) is True
 
 
 def test_a_conversation_with_no_rows_is_not_unanswered():

memhub/docs/MEMORY_EVENT_TIME_SCHEDULING_PLAN.md

@@ -15,7 +15,7 @@ fixed ten-minute event-time window.
    - Parse native Harness timestamps.
    - Parse Cursor `<timestamp>` envelopes before stripping them.
    - Mark fallback quality explicitly.
-   - Derive each chunk's time from its person-authored user opener.
+   - Derive each chunk's time from its protocol `user` opener.
 
 2. **Durable schema**
    - Add `event_at` and `time_quality` to `memory_write_jobs`.
@@ -25,7 +25,8 @@ fixed ten-minute event-time window.
 3. **Windowed Drain**
    - Anchor at the oldest eligible pending event.
    - Admit only `[anchor, anchor + 10 minutes)`.
-   - Keep one running job per Harness and FIFO by `queue_seq`.
+   - Keep one running job per Harness; order globally by `event_at`.
+   - Use `queue_seq` only to break equal-time ties within one Session.
    - Advance only after the current window has no eligible pending/running
      work.
 

memhub/docs/contracts/memory-event-time-scheduling-v1.md

@@ -16,12 +16,12 @@ The row stores:
 
 - `event_at`: UTC event time used for scheduling;
 - `time_quality`: `exact`, `anchored`, `session`, or `approximate`;
-- `queue_seq`: the existing monotonic tie-breaker and replay order;
+- `queue_seq`: a deterministic tie-breaker only when event times are equal;
 - `session_key` and `source_key`: the existing session/Harness identities.
 
 Time selection is:
 
-1. the person-authored user turn timestamp;
+1. the protocol `user` turn timestamp;
 2. the nearest session timestamp anchor;
 3. the session/file timestamp;
 4. `imported_at` or write creation time.
@@ -58,8 +58,10 @@ Within one window:
 - different Harnesses may run concurrently, up to the existing concurrency
   limit;
 - one Harness still has at most one running job;
-- earlier `queue_seq` values within a Harness block later values;
-- jobs are selected by `(event_at, queue_seq, job_id)`;
+- each Session is ordered by `(event_at, queue_seq)`;
+- different Sessions from one Harness never block each other's event order;
+- `source_key` limits one running job per Harness but does not define FIFO;
+- jobs are selected globally by `(event_at, queue_seq, job_id)`;
 - `live` affects service admission only and does not jump the time window.
 
 When the active window is empty, the next oldest eligible pending job becomes
@@ -91,16 +93,31 @@ batch are not held.
 - Claimed but not invoked: return to pending without consuming an attempt.
 - Already invoked: the current provider call may finish; it is never blindly
   cancelled or retried.
-- Unknown outcomes retain the existing reconcile-before-retry contract.
+- Unknown outcomes retain their complete frozen input, never block another
+  Session or Harness, and follow the bounded reconcile-before-retry contract.
+
+## Immutable job input and branches
+
+Every new `memory_write_jobs.payload_json` stores the complete immutable
+Evidence records used by the write, not only references to another table.
+Execution and unknown reconciliation use that snapshot. Terminal jobs retain
+it as audit data.
+
+When a Harness rewinds and continues from an earlier point, already stored
+Evidence and jobs are not deleted. Shared unchanged messages deduplicate;
+changed or new branch records append new jobs. A later revision of a reused
+host message id receives a new write id while the earlier job keeps its
+original content.
 
 ## Upgrade compatibility
 
 Legacy jobs are assigned:
 
-1. the latest person-user Evidence time linked to the job;
+1. the latest `role=user` Evidence time linked to the job;
 2. otherwise the earliest linked Evidence time;
 3. otherwise the original job creation time.
 
 They are marked `approximate` and retain `queue_seq` as a deterministic
-tie-breaker. No payload, Evidence identity, or idempotency key is rewritten.
+tie-breaker. Old id-only payloads remain readable; all newly submitted jobs
+are self-contained.