Stop repeated goal/store work without stale goal publication

src/agent_comms/acp.py

@@ -1,3 +1,3 @@
 """Agent communications — ACP server on the official protocol library.
 
 Implements an Agent Client Protocol agent so ACP clients (Toad, Zed, VS Code)
@@ -20,7 +20,6 @@
 import asyncio
 import json
 import os
-import re
 import sys
 import time
 from collections.abc import AsyncIterator, Iterator, Sequence
@@ -60,8 +59,6 @@
     manual_compaction_bridge,
 )
 from .acp_extension import (
-    CompactRequest,
-    decode_request,
     encode_updates,
 )
 from .acp_failure import NativeTurnError
@@ -79,7 +76,7 @@
 from .coordination_database import CoordinationStore
 from .cursor_owner import CursorOwner
 from .cursor_publication import CursorPublication
-from .field_codec import ClaimId, FieldCodec, MalformedRecord, RootId
+from .field_codec import AcpSessionId, ClaimId, FieldCodec, RootId
 from .image_inputs import prompt_images
 from .input_drain import InputDrain
 from .config_options import ConfigOptions
@@ -117,17 +114,17 @@
 class SdkSessionConnection(AgentSideConnection, Controller, SessionUpdateSink):
     """The SDK connection and controller publish one Core acceptance answer."""
 
-    async def publish_update(self, *, session_id: str, update: BaseModel) -> bool:
+    async def publish_update(self, *, session_id: AcpSessionId, update: BaseModel) -> bool:
         await self.session_update(session_id=session_id, update=update)
         return True
 
 
-    def require_current(self, authority: AcpSelectedWriteAuthority, session_id: str) -> None:
+    def require_current(self, authority: AcpSelectedWriteAuthority, session_id: AcpSessionId) -> None:
         if self is not authority.agent.sessions.client:
             raise IdentityConflict("Selected write ACP controller changed")
 
     async def request_ui_permission(
-        self, runner: TurnRunner, session_id: str,
+        self, runner: TurnRunner, session_id: AcpSessionId,
         request: tuple[ToolCallUpdate, list[PermissionOption]],
     ) -> dict[str, object] | None:
         tool_call, options = request
@@ -143,7 +140,7 @@
             )
         return None
 
-    def departed(self, runtime: RuntimeServer, session_id: str) -> bool:
+    def departed(self, runtime: RuntimeServer, session_id: AcpSessionId) -> bool:
         return False
 
 
@@ -198,9 +195,10 @@
         # remains attached. Never restore this binding after a process crash.
         self._selected_write_controllers: dict[tuple[str, int], tuple[ClaimId, Controller]] = {}
 
-    def _create_runtime_proxy(self, thread: Thread, session_id: str) -> RuntimeProxy:
+    def _create_runtime_proxy(self, thread: Thread, session_id: AcpSessionId) -> RuntimeProxy:
         return RuntimeProxy(
-            self, session_id, socket_path(self._comms.root, thread.require_process().pid),
+            self, thread.incarnation, session_id,
+            socket_path(self._comms.root, thread.require_process().pid),
         )
 
     # Private attachment ABI consumed by runtime/compaction. State exists only
@@ -287,19 +285,11 @@
         return await self.receive_prompt(request)
 
     async def receive_prompt(self, request: PromptRequest) -> PromptResponse:
-        try:
-            command = decode_request(request.field_meta)
-        except MalformedRecord as error:
-            raise RequestError.invalid_params({"reason": str(error)}) from error
-        # Literal /compact is the external ACP text command; ours uses a record.
-        text = self._prompt_text(request).strip()
-        if re.match(r"^/compact(?:\s|$)", text):
-            self._require_compaction_text(request)
-            command = CompactRequest(text[len("/compact") :].strip())
         with self.refusals():
-            return await AcpRequestConsumer(self, request).run(command)
+            session_id = FieldCodec.decode(AcpSessionId, request.session_id)
+            return await AcpRequestConsumer(self, request, session_id).run()
 
-    async def _compact_request(self, session_id: str, instructions: str | None) -> PromptResponse:
+    async def _compact_request(self, session_id: AcpSessionId, instructions: str | None) -> PromptResponse:
         # The owner process's bridge alone owns the turn lock.
         if session_id in self.sessions.proxies:
             outcome = FieldCodec.decode(
@@ -314,7 +304,7 @@
             )
         return manual_compaction_bridge.CompactionReply().reply(outcome)
 
-    async def _selected_write_request(self, session_id: str, request) -> PromptResponse:
+    async def _selected_write_request(self, session_id: AcpSessionId, request) -> PromptResponse:
         owner = self.sessions.require(session_id)
         controller = self._runtime.controller.get()
         attached = controller.attached(self.sessions.client) if controller is not None else None
@@ -341,9 +331,10 @@
         return PromptResponse(stop_reason="end_turn", field_meta=encode_updates(acknowledgement))
 
     async def _prompt_request(
-        self, request: PromptRequest, command: acp_extension.PromptRequest
+        self, ingress: AcpRequestConsumer, command: acp_extension.PromptRequest
     ) -> PromptResponse:
-        session_id = request.session_id
+        session_id = ingress.session_id
+        request = ingress.request
         text = self._prompt_text(request)
         display_text = command.user_text or text
         try:
@@ -378,7 +369,8 @@
             )
             try:
                 return await self.turns.prompt_owned(
-                    request,
+                    session_id,
+                    text=text,
                     images=images,
                     display_text=display_text,
                     input_id=command.input_id,
@@ -400,7 +392,7 @@
             )
 
     async def emit_session_identity(
-        self, session_id: str, name: str, client: SessionUpdateSink,
+        self, session_id: AcpSessionId, name: str, client: SessionUpdateSink,
     ) -> None:
         """Let a subscriber identify its owner before potentially long replay."""
         await client.publish_update(
@@ -420,24 +412,24 @@
 
     async def cancel(self, session_id: str, **kwargs: JsonValue) -> None:
         with self.refusals():
-            return await self.turns.cancel(session_id, **kwargs)
+            return await self.turns.cancel(FieldCodec.decode(AcpSessionId, session_id), **kwargs)
 
     async def close_session(self, session_id: str, **kwargs: JsonValue):
         with self.refusals():
-            return await self.sessions.close_session(session_id)
+            return await self.sessions.close_session(FieldCodec.decode(AcpSessionId, session_id))
 
     async def set_config_option(
         self, config_id: str, session_id: str, value: str | bool, **kwargs: JsonValue
     ) -> SetSessionConfigOptionResponse:
         with self.refusals():
-            return await self.sessions.config.set_option(config_id, session_id, value)
+            return await self.sessions.config.set_option(config_id, FieldCodec.decode(AcpSessionId, session_id), value)
 
     async def authenticate(self, method_id: str, **kwargs: JsonValue) -> None:
         raise RequestError.auth_required({"reason": "agent-comms requires no authentication"})
 
     # ─── Helpers ─────────────────────────────────────────────────────────────
 
-    async def _drain_private_nk(self, session_id: str, wire_root_id: RootId) -> int:
+    async def _drain_private_nk(self, session_id: AcpSessionId, wire_root_id: RootId) -> int:
         """Run a selected private wake for this ACP session.
 
         This explicitly configured path reuses the reviewed one-shot native

src/agent_comms/acp_extension.py

@@ -22,13 +22,14 @@
     FailureSource,
     UnrecordedFailureSource,
 )
-from .agent_backend import SendNow, TurnInbox
+from .turn_inbox import SendNow, TurnInbox
 from .agent_events import CompactionEvent
 from .child_process import ProcessIdentity
 from .context_reading import ContextReading
 from .declared_family import DeclaredFamily
 from .external_transcript import DraftRow
 from .field_codec import (
+    AcpSessionId,
     AssignmentId,
     ClaimId,
     AcceptsNull,
@@ -435,7 +436,7 @@ def projection(self, queued, restored) -> QueueProjection:
     def validate_scope(self, projection: QueueProjection) -> None:
         projection.without_scope()
 
-    def start_revision(self, inputs: InputDrain, session_id: str) -> int | None:
+    def start_revision(self, inputs: InputDrain, session_id: AcpSessionId) -> int | None:
         return None
 
     def accepts_request(self, projection: QueueProjection) -> bool:
@@ -474,7 +475,7 @@ def projection(self, queued, restored) -> QueueProjection:
     def validate_scope(self, projection: QueueProjection) -> None:
         pass
 
-    def start_revision(self, inputs: InputDrain, session_id: str) -> int:
+    def start_revision(self, inputs: InputDrain, session_id: AcpSessionId) -> int:
         return inputs.next_revision(session_id)
 
     def accepts_request(self, projection: QueueProjection) -> bool:
@@ -753,7 +754,7 @@ def draft_text(self) -> str | None:
     @abstractmethod
     def enqueue_control(self, inbox: TurnInbox, input_id: InputId) -> None: ...
 
-    async def publish_acceptance(self, inputs: InputDrain, session_id: str) -> None:
+    async def publish_acceptance(self, inputs: InputDrain, session_id: AcpSessionId) -> None:
         await inputs.emit_queue_state(session_id)
 
 

src/agent_comms/acp_request_consumer.py

@@ -1,5 +1,7 @@
 """Declared owner commands at the ACP boundary; shared behavior is inherited."""
 
+import re
+
 from metaclass_registry.mro_dispatch import (
     MroDispatch,
     handles,
@@ -14,70 +16,82 @@
     CompactRequest,
     SelectedWriteRequest,
     SendNowRequest,
+    decode_request,
     encode_request,
 )
+from .field_codec import AcpSessionId, MalformedRecord
 
 
 class AcpRequestConsumer(MroDispatch):
-
-    def __init__(self, agent, request: PromptRequest):
-        self.agent, self.request = agent, request
+    def __init__(self, agent, request: PromptRequest, session_id: AcpSessionId):
+        self.agent, self.request, self.session_id = agent, request, session_id
         self.response = None
 
-    async def run(self, request):
+    async def run(self):
+        try:
+            request = decode_request(self.request.field_meta)
+        except MalformedRecord as error:
+            raise Refused.invalid_params({"reason": str(error)}) from error
+        # Literal /compact is the external ACP text command; ours uses a record.
+        text = self.agent._prompt_text(self.request).strip()
+        if re.match(r"^/compact(?:\s|$)", text):
+            self.agent._require_compaction_text(self.request)
+            request = CompactRequest(text[len("/compact") :].strip())
         await self.dispatch(request)
         if self.response is None:
             raise RuntimeError("Declared ACP request has no response handler")
         return self.response
 
     async def forward(self, request):
         forwarded = self.request.model_copy(
-            update={"field_meta": {**(self.request.field_meta or {}), **encode_request(request)}}
+            update={
+                "session_id": self.session_id,
+                "field_meta": {**(self.request.field_meta or {}), **encode_request(request)},
+            }
         )
-        result = await self.agent.sessions.proxies[self.request.session_id].request(
+        result = await self.agent.sessions.proxies[self.session_id].request(
             "prompt", request=forwarded
         )
         return PromptResponse.model_validate(result)
 
     @handles(acp_extension.PromptRequest)
     async def prompt_request(self, request):
-        if self.request.session_id in self.agent.sessions.proxies:
+        if self.session_id in self.agent.sessions.proxies:
             self.response = await self.forward(request)
             return
-        self.response = await self.agent._prompt_request(self.request, request)
+        self.response = await self.agent._prompt_request(self, request)
 
     @handles(CompactRequest)
     async def compact(self, request):
         self.agent._require_compaction_text(self.request)
-        if self.agent._prompt_text(self.request).strip() and not self.agent._prompt_text(
-            self.request
-        ).strip().startswith("/compact"):
+        if (
+            self.agent._prompt_text(self.request).strip()
+            and not self.agent._prompt_text(self.request).strip().startswith("/compact")
+        ):
             raise Refused.invalid_params({"reason": "Compaction metadata requires blank text"})
-        self.response = await self.agent._compact_request(
-            self.request.session_id, request.instructions
-        )
+        self.response = await self.agent._compact_request(self.session_id, request.instructions)
 
     @handles(SelectedWriteRequest)
     async def selected_write(self, request):
         if self.request.prompt:
             raise Refused.invalid_params({"reason": "Selected write requires no prompt"})
-        if self.request.session_id in self.agent.sessions.proxies:
+        if self.session_id in self.agent.sessions.proxies:
             self.response = await self.forward(request)
             return
-        self.response = await self.agent._selected_write_request(self.request.session_id, request)
+        self.response = await self.agent._selected_write_request(self.session_id, request)
 
     @handles(ClearQueueRequest)
     async def clear(self, request):
-        if self.request.session_id in self.agent.sessions.proxies:
-            await self.agent.sessions.proxies[self.request.session_id].request("clear_queue")
+        if self.session_id in self.agent.sessions.proxies:
+            await self.agent.sessions.proxies[self.session_id].request("clear_queue")
         else:
-            await self.agent.inputs.clear_queued_inputs(self.request.session_id)
+            await self.agent.inputs.clear_queued_inputs(self.session_id)
         self.response = PromptResponse(stop_reason="end_turn")
 
     @handles(SendNowRequest)
     async def send_now(self, request):
-        if self.request.session_id in self.agent.sessions.proxies:
+        if self.session_id in self.agent.sessions.proxies:
             self.response = await self.forward(request)
             return
-        await self.agent.inputs.send_now(self.request.session_id)
+        await self.agent.inputs.send_now(self.session_id)
         self.response = PromptResponse(stop_reason="end_turn")

src/agent_comms/agent_activity.py

@@ -173,14 +173,17 @@ def observe_recipients(
         snapshot: RegistrySnapshot,
     ) -> Mapping[ThreadLookup, RecipientActivity]:
         """Acquire one bounded window's recipients from the original sends."""
-        observations: dict[ThreadLookup, RecipientActivity] = {}
+        selected = {}
         for original, recipient in recipients:
             lookup = recipient.recipient_lookup
-            if lookup in observations:
-                continue
-            selected = snapshot.recipient(recipient, original)
-            observations[lookup] = selected.activity(self, snapshot)
-        return observations
+            if lookup not in selected:
+                selected[lookup] = snapshot.recipient(recipient, original)
+        captured = self.all_activity(
+            (thread for recipient in selected.values() for thread in recipient.activity_threads()),
+            snapshot=snapshot,
+        )
+        return {lookup: recipient.activity(captured, snapshot)
+                for lookup, recipient in selected.items()}
 
     def _emit_activity(self, activity: Activity) -> None:
         snapshot = self.registry.snapshot()