Preserve inbox analysis across coding handoffs

docs/engineering/codex_dgx_worker.md

@@ -433,6 +433,15 @@ existing canonical task fingerprint keyed by source message and configured
 participants, followed by assignment/text read-back and a provenance comment.
 A retry reuses that assignment without resetting an already progressed task.
 
+New and resumed coding follow-ups also carry `agent_analysis`: the inbox answer
+bound to its source message, configured agent and organisation. It remains
+separate from the exact source instructions and never grants deployment or
+assignment authority. Coalesced follow-ups retain each analysis with its own
+source; retries reuse the existing handoff. Older retained follow-ups without
+this optional context remain valid. Completing a review is distinct from
+delivering the implementation it recommends; retain unfinished implementation
+in an explicit native task instead of reporting queue acceptance as completion.
+
 Recent direct-message context selects the newest 30 messages in the configured
 organisation at or before the source timestamp, then presents them chronologically.
 The context records its time boundary and whether older messages were omitted.

scripts/codex_von_inbox.py

@@ -71,6 +71,17 @@ def task_followup(config, task_id):
     return json.loads(path.read_text()) if path.exists() else None
 
 
+def analysis_context(config, message, result):
+    """Carry the inbox judgement separately from the delegator's instructions."""
+    return {
+        "source_message_id": message["message_id"],
+        "actor_id": config["agent_id"],
+        "organisation_id": config["organisation_id"],
+        "content": result["answer"],
+        "authority": "Agent analysis of the source request; not additional authority.",
+    }
+
+
 def already_answered(config, message_id):
     path = state_path(config, message_id)
     return path.exists() and json.loads(path.read_text()).get("phase") == "done"
@@ -488,6 +499,7 @@ def create_assignment(config, api, state, path):
         "message_id": message["message_id"],
         "content": message["content"],
         "deployment_requested": state["result"]["deployment_requested"],
+        "agent_analysis": [analysis_context(config, message, state["result"])],
     }
     if not task_followup(config, task_id):
         write_json(followup_path(config, task_id), followup)
@@ -549,6 +561,7 @@ def finish(config, api, state, path):
                 "message_id": message["message_id"],
                 "content": message["content"],
                 "deployment_requested": result["deployment_requested"],
+                "agent_analysis": [analysis_context(config, message, result)],
                 **({"retry": state["retry_intent"]} if state["retry_intent"] else {}),
                 **(
                     {"attachments": message["attachments"]}
@@ -583,6 +596,10 @@ def finish(config, api, state, path):
                         *prior.get("message_ids", [prior["message_id"]]),
                         message["message_id"],
                     ]
+                    followup["agent_analysis"] = [
+                        *prior.get("agent_analysis", []),
+                        *followup["agent_analysis"],
+                    ]
                     if prior.get("attachments") or followup.get("attachments"):
                         followup["attachments"] = [
                             *prior.get("attachments", []),

scripts/codex_von_worker_prompt.md

@@ -57,6 +57,12 @@ explicit deployment instruction and deployment_requested value; an old deploy
 request does not carry forward. Otherwise preserve the initial-task deployment
 rule below.
 
+When followup.agent_analysis is present, use its source-bound findings and
+proposed work as context from the prior inbox run. It is the named agent's
+analysis, not a delegator instruction or an authority grant. Check it against
+the source request and current evidence; preserve any unresolved implementation
+work separately from completion of a bounded review.
+
 The controller sends your result or question to Michael in Von, records the
 worktree and evidence on the task, and waits for a task-thread reply, task
 comment, or requeue when you need input. Do not send messages yourself or

tests/backend/test_codex_von_inbox.py

@@ -4,9 +4,17 @@
 from types import SimpleNamespace
 
 import pytest
+import mongomock
 
 from scripts import codex_von_inbox as inbox
 from scripts import codex_von_worker as worker
+from src.backend.services import task_dispatch_authority_service as dispatch
+
+
+@pytest.fixture(autouse=True)
+def isolated_dispatch_store(monkeypatch):
+    collection = mongomock.MongoClient().test.dispatch
+    monkeypatch.setattr(dispatch, "_collection", lambda: collection)
 
 
 @pytest.fixture
@@ -119,6 +127,15 @@ def launch(config, state, fd):
             "message_id": "#V#question",
             "content": message["content"],
             "deployment_requested": deploy,
+            "agent_analysis": [
+                {
+                    "source_message_id": "#V#question",
+                    "actor_id": "#V#agent",
+                    "organisation_id": "#V#org",
+                    "content": "The tray was deployed.",
+                    "authority": "Agent analysis of the source request; not additional authority.",
+                }
+            ],
         }
     else:
         assert inbox.task_followup(config, "#V#task") is None
@@ -204,6 +221,10 @@ def report():
     assert "Move the send controls" in followup["content"]
     assert "compact identity labels" in followup["content"]
     assert followup["message_ids"] == ["#V#question", "#V#second_question"]
+    assert [row["source_message_id"] for row in followup["agent_analysis"]] == [
+        "#V#question",
+        "#V#second_question",
+    ]
     assert task["status"] == "pending" and len(comments) == 2
     assert transitions == ["pending"]
 
@@ -455,6 +476,9 @@ def create_task(**kw):
     assert created[0]["agent_creation_request_id"] == message["message_id"]
     assert len(comments) == 1
     assert inbox.task_followup(config, "#V#new")["deployment_requested"] is False
+    assert inbox.task_followup(config, "#V#new")["agent_analysis"][0]["content"] == (
+        "I will implement the requested change."
+    )
     assert not inbox.tick(config, api, 0)
 
 

tests/backend/test_codex_von_inbox_canonical_replay.py

@@ -16,9 +16,10 @@
 from src.backend.security.access_control import override_current_actor
 from src.backend.services import message_service as messages
 from src.backend.services import organisation_membership_service as membership
+from src.backend.services import task_dispatch_authority_service as dispatch
 
 
-@pytest.mark.parametrize("action", ["reply", "create_task"])
+@pytest.mark.parametrize("action", ["reply", "create_task", "resume_task"])
 def test_canonical_delivery_assignment_and_crash_recovery(
     native_store, tmp_path, monkeypatch, action
 ):
@@ -33,6 +34,7 @@ def test_canonical_delivery_assignment_and_crash_recovery(
         "model": "gpt-6-astra",
         "model_reasoning_effort": "xhigh",
     }
+    monkeypatch.setattr(dispatch, "get_db", lambda: native_store)
     monkeypatch.setattr(
         messages, "get_concepts_collection", lambda: native_store.concepts
     )
@@ -86,7 +88,11 @@ def test_canonical_delivery_assignment_and_crash_recovery(
             content=(
                 captured["message"]["content"]
                 if action == "reply"
-                else "Implement the new export control and deploy it. Use gpt-6-astra with xhigh reasoning. Fixture request."
+                else (
+                    f"Resume {old['task_concept_id']}: implement the export control. No deployment."
+                    if action == "resume_task"
+                    else "Implement the new export control and deploy it. Use gpt-6-astra with xhigh reasoning. Fixture request."
+                )
             ),
         )
         response = {
@@ -95,17 +101,17 @@ def test_canonical_delivery_assignment_and_crash_recovery(
                 if action == "reply"
                 else "I will implement the new export control."
             ),
-            "task_id": "",
+            "task_id": old["task_concept_id"] if action == "resume_task" else "",
             "action": action,
             "deployment_requested": action == "create_task",
             "new_task": (
-                None
-                if action == "reply"
-                else {
+                {
                     "title": "New export control",
                     "requested_model": "gpt-6-astra",
                     "requested_reasoning_effort": "xhigh",
                 }
+                if action == "create_task"
+                else None
             ),
         }
         calls = []
@@ -145,7 +151,9 @@ def crash_after_delivery(path, value):
         )
         assert response["answer"] in delivered["content"]
         assert delivered["reply_to_id"] == source["concept_id"]
-        assert api.task(old["task_concept_id"])["status"] == "completed"
+        assert api.task(old["task_concept_id"])["status"] == (
+            "pending" if action == "resume_task" else "completed"
+        )
         tasks = api.tasks.search_tasks(
             assignee_concept_id="#V#worker",
             created_by_concept_id="#V#alice",
@@ -173,5 +181,24 @@ def crash_after_delivery(path, value):
                 len(api.tasks.list_task_comments(new["task_concept_id"])["comments"])
                 == 1
             )
-        else:
+        elif action == "reply":
             assert not list((tmp_path / "followups").glob("*.json"))
+        if action != "reply":
+            task_id = checkpoint.get("created_task_id", old["task_concept_id"])
+            coding_input = api.inputs(api.task(task_id))
+            assert (
+                coding_input["followup"]["content"]
+                == source["concept_data"]["content_fallback"]
+            )
+            assert coding_input["followup"]["agent_analysis"] == [
+                {
+                    "source_message_id": source["concept_id"],
+                    "actor_id": "#V#worker",
+                    "organisation_id": "#V#lab",
+                    "content": response["answer"],
+                    "authority": "Agent analysis of the source request; not additional authority.",
+                }
+            ]
+            # The agent's canonical note is deliberately not promoted to an
+            # instruction. Its analysis still reaches the next coding run once.
+            assert coding_input["comments"] == []