Represent Claude compaction as connected bridge presence

pyproject.toml

@@ -58,6 +58,8 @@ agent-comms-codex-bridge = "agent_comms.cli_bridges.codex:main"
 agent-comms-opencode-bridge = "agent_comms.cli_bridges.opencode:main"
 agent-comms-commander = "agent_comms.commander.cli:main"
 agent-comms-watch-guard = "agent_comms_hooks.watch:main"
+agent-comms-claude-precompact = "agent_comms_hooks.compaction:starting"
+agent-comms-claude-compacted = "agent_comms_hooks.compaction:finished"
 agent-comms-connector = "agent_comms.connector:main"
 opencode-comms = "agent_comms.cli_bridges.opencode_launch:main"
 

src/agent_comms/cli_bridges/shared.py

@@ -283,15 +283,16 @@ async def run(self, bridge: OperationBridge) -> None:
         # name the re-arm command once the lease lapses.
         record = WatchRecord(bridge.binding.bus, bridge.binding.root,
                              (Path(sys.argv[0]).name, *sys.argv[1:]))
-        with WatchClaim(bridge.binding.thread).held(record):
-            from agent_comms.registration import Registration
-
-            registry = await asyncio.to_thread(
-                Registration, Path(bridge.binding.root) / 'registry.json')
-            # Retry only endpoint acquisition, never the running bridge's delivery work.
-            with ExitStack() as endpoint:
-                await bridge.comms.bus.when_open(lambda: endpoint.enter_context(
-                    registry.external_process(bridge.binding.bus, ProcessIdentity.capture(os.getpid()))))
+        from agent_comms.registration import Registration
+
+        registry = await asyncio.to_thread(
+            Registration, Path(bridge.binding.root) / 'registry.json')
+        # The live claim names an already registered endpoint, never a process
+        # still waiting to attach. Retry only acquisition, not delivery work.
+        with ExitStack() as endpoint:
+            await bridge.comms.bus.when_open(lambda: endpoint.enter_context(
+                registry.external_process(bridge.binding.bus, ProcessIdentity.capture(os.getpid()))))
+            with WatchClaim(bridge.binding.thread).held(record):
                 async with asyncio.TaskGroup() as tasks:
                     self.start(bridge, tasks)
 

src/agent_comms/registration.py

@@ -18,7 +18,7 @@
 from .registry_store import RegistryEdit, RegistryEntryRevision, RegistryStore
 from .routing import TurnRouting
 from .store_files import SharedLock, _store_lock
-from .thread_identity import GenerationCounter, TurnId
+from .thread_identity import GenerationCounter, OwnerIdentity, TurnId
 from .thread_status import (
     ConnectedThreadStatus, NoBridgeThreadStatus, RunningThreadStatus, ThreadStatus,
 )
@@ -104,6 +104,18 @@ def external_process(self, name: str, process: ProcessIdentity):
                         replace(entry.thread, process_identity=NoProcess()), NoBridgeThreadStatus(),
                         new_owner=False))
 
+    def observe_external_status(
+        self, owner: OwnerIdentity, process: ProcessIdentity, status: ThreadStatus,
+    ) -> None:
+        """An observation changes only the bridge lease it actually witnessed."""
+        with self.store.editing() as edit:
+            entry = edit.document.entry_if_present(owner.incarnation.name)
+            if entry is None or not entry.owned_by(owner, process) or not process.alive():
+                return
+            entry.thread.execution.require_external()
+            edit.document.statuses[entry.thread.name] = status.with_bridge()
+            edit.commit()
+
     def attach_native_session(self, original: RegistryOwner, session_file: str) -> RegistryOwner:
         """Publish an observed native source under its original executable lease.
 

src/agent_comms/thread_execution.py

@@ -8,7 +8,7 @@
 from .declared_family import DeclaredFamily
 from .child_process import ProcessState
 from .errors import RelationViolationError
-from .thread_status import ConnectedThreadStatus, NoBridgeThreadStatus, ThreadStatus
+from .thread_status import NoBridgeThreadStatus, ThreadStatus
 
 if TYPE_CHECKING:
     from .threads import Thread
@@ -165,7 +165,7 @@ def require_external(cls) -> None:
     def registered_status(cls, thread: Thread, requested: ThreadStatus) -> ThreadStatus:
         """Core launches no process here: the inbox is connected while its bridge lives."""
         if requested.active and thread.process_alive:
-            return ConnectedThreadStatus()
+            return requested.with_bridge()
         return NoBridgeThreadStatus()
 
     @classmethod

src/agent_comms/thread_status.py

@@ -10,7 +10,10 @@
 from .declared_family import DeclaredFamily
 from .errors import RelationViolationError
 from .field_codec import FieldCodec, TextRepresentation
-from .thread_presentation import ThreadOwnerBinding, ThreadPresentation, InactiveThreadPresentationStatus
+from .thread_presentation import (
+    ThreadOwnerBinding, ThreadPresentation, InactiveThreadPresentationStatus,
+    WorkingThreadPresentationStatus,
+)
 
 if TYPE_CHECKING:
     from .activity import ObservedActivity
@@ -78,6 +81,10 @@ def native_presentation(
     ) -> ThreadPresentation:
         return ordinary
 
+    def with_bridge(self) -> ThreadStatus:
+        """An attached bridge supplies connected presence, not a native lifecycle."""
+        return ConnectedThreadStatus()
+
     def external_presentation(
         self, process: ProcessState, ordinary: ThreadPresentation,
     ) -> ThreadPresentation:
@@ -167,6 +174,19 @@ def after_heartbeat(self, name: str) -> ThreadStatus:
 class ConnectedThreadStatus(CliInboxPresence, RunningThreadStatus):
     """A CLI inbox whose bridge process is watching it."""
 
+    def with_bridge(self) -> ThreadStatus:
+        return self
+
+
+class CompactingThreadStatus(ConnectedThreadStatus):
+    """The bridge is still watching while its CLI compacts context."""
+
+    receiving = False
+
+    def presentation(self, title: str, activity: ObservedActivity) -> ThreadPresentation:
+        return ThreadPresentation(title, "⌛", "Compacting context",
+                                  status=WorkingThreadPresentationStatus())
+
 
 class NoBridgeThreadStatus(CliInboxPresence, StoppedThreadStatus):
     """A CLI inbox with no bridge attached; its messages wait unread.

src/agent_comms_hooks/compaction.py

@@ -0,0 +1,50 @@
+"""Claude's real compaction hooks observe the connected watch, never create one."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from agent_comms_hooks import ClaudeHook, ClaudeHookProgram
+from agent_comms_hooks.watch import WatchClaim
+
+
+class BeforeCompaction(ClaudeHookProgram):
+    events = ('PreCompact',)
+
+    @classmethod
+    def command(cls) -> str:
+        return cls.console_script(f'{starting.__module__}:{starting.__name__}')
+
+
+class AfterCompaction(ClaudeHookProgram):
+    events = ('SessionStart',)
+
+    @classmethod
+    def command(cls) -> str:
+        return cls.console_script(f'{finished.__module__}:{finished.__name__}')
+
+    @classmethod
+    def entry(cls) -> dict[str, object]:
+        return {'matcher': 'compact', **super().entry()}
+
+
+@dataclass(frozen=True)
+class SessionStartHook(ClaudeHook):
+    source: str
+
+
+def starting() -> None:
+    hook = ClaudeHook.read()
+    from agent_comms.thread_status import CompactingThreadStatus
+
+    WatchClaim(hook.session_id).state().observe_compaction(CompactingThreadStatus())
+
+
+def finished() -> None:
+    hook = SessionStartHook.read()
+    if hook.source != 'compact':
+        return
+    from agent_comms.thread_status import ConnectedThreadStatus
+
+    # SessionStart(source=compact) is Claude's first post-compaction observation,
+    # including a manual /compact with no subsequent model turn or file update.
+    WatchClaim(hook.session_id).state().observe_compaction(ConnectedThreadStatus())

src/agent_comms_hooks/watch.py

@@ -15,6 +15,10 @@
 from contextlib import contextmanager
 from dataclasses import asdict, dataclass
 from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from agent_comms.thread_status import ThreadStatus
 
 from agent_comms_hooks import ClaudeHook, ClaudeHookProgram
 
@@ -53,14 +57,17 @@ def state(self) -> WatchState:
             try:
                 fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB)
             except BlockingIOError:
-                return LiveWatch()
+                return LiveWatch(self, WatchRecord(**json.load(lock)))
             text = lock.read()
         if not text:
             return UnclaimedWatch()
         return LapsedWatch(self, WatchRecord(**json.loads(text)))
 
 
 class WatchState:
+    def observe_compaction(self, status: ThreadStatus) -> None:
+        """An absent watch never acquires bridge presence from a hook."""
+
     def report(self, event: str) -> None:
         """Silent unless the conversation is meant to be watched and is not."""
 
@@ -69,8 +76,25 @@ class UnclaimedWatch(WatchState):
     pass
 
 
+@dataclass(frozen=True)
 class LiveWatch(WatchState):
-    pass
+    claim: WatchClaim
+    record: WatchRecord
+
+    def observe_compaction(self, status: ThreadStatus) -> None:
+        # Hooks on every prompt keep their stdlib-only import path. Only an
+        # actual compaction observation needs the registry transaction owner.
+        from agent_comms.registration import Registration
+
+        registry = Registration(Path(self.record.root) / 'registry.json')
+        original = registry.entry(self.record.bus).entry
+        # The watch must still be the one read before importing Core. Its claim
+        # is held inside its registered endpoint lifetime; the transaction then
+        # fences that exact incarnation, generation and process, not a name.
+        if self.claim.state() == self:
+            registry.observe_external_status(
+                original.owner_identity, original.thread.require_process(), status)
+
 
 
 @dataclass(frozen=True)

tests/test_thread_status.py

@@ -17,6 +17,9 @@
 from agent_comms.registration import Registration
 from agent_comms.thread_status import (
     ArchivedThreadStatus,
+    CompactingThreadStatus,
+    ConnectedThreadStatus,
+    NoBridgeThreadStatus,
     DeletingThreadStatus,
     IdleThreadStatus,
     RunningThreadStatus,
@@ -26,6 +29,47 @@
 from agent_comms.threads import Thread
 
 
+def test_connected_compaction_preserves_endpoint_and_presentation(tmp_path, monkeypatch):
+    from agent_comms.thread_execution import ExternalThreadExecution
+    from agent_comms_hooks.watch import WatchClaim, WatchRecord
+
+    monkeypatch.setenv('HOME', str(tmp_path))
+    comms = wire(tmp_path / 'bus')
+    registry = comms.registry
+    registry.declare(registry.new_thread(
+        'watched', frozenset(), str(tmp_path), execution=ExternalThreadExecution))
+    process = ProcessIdentity.capture(os.getpid())
+    claim = WatchClaim('session')
+    record = WatchRecord('watched', str(comms.root), ('watch',))
+    with registry.external_process('watched', process):
+        with claim.held(record):
+            original = registry.entry('watched').entry
+            watch = claim.state()
+            watch.observe_compaction(CompactingThreadStatus())
+            compacting = registry.snapshot()
+            assert compacting.statuses['watched'] == CompactingThreadStatus()
+            assert compacting.threads['watched'].process_identity == process
+            assert compacting.owner_identity('watched') == original.owner_identity
+            registry.heartbeat('watched')
+            registry.register(replace(registry.require('watched'), title='Current title'), registry.status('watched'))
+            shown = comms.views.thread_presentations(('watched',)).require('watched')
+            assert shown.summary == 'Compacting context' and shown.status.working
+            assert shown.title == 'Current title'
+            assert registry.status('watched') == CompactingThreadStatus()
+            watch.observe_compaction(ConnectedThreadStatus())
+            assert registry.status('watched') == ConnectedThreadStatus()
+            assert registry.entry('watched').entry.owner_identity == original.owner_identity
+            watch.observe_compaction(CompactingThreadStatus())
+    assert registry.status('watched') == NoBridgeThreadStatus()
+    watch.observe_compaction(CompactingThreadStatus())
+    assert registry.status('watched') == NoBridgeThreadStatus()
+    with registry.external_process('watched', process):
+        # An earlier bridge's delayed observation cannot alter the new lease,
+        # even when the process is the same and only the generation changed.
+        registry.observe_external_status(original.owner_identity, process, CompactingThreadStatus())
+        assert registry.status('watched') == ConnectedThreadStatus()
+
+
 def test_captured_owner_presentation_survives_exit_until_fresh_acquisition(tmp_path):
     comms = wire(tmp_path)
     child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])

tests/test_watch_guard.py

@@ -40,7 +40,10 @@ def hook(event='PostToolUse', session='session'):
             for event, entries in hooks.items()} == {
         'PreToolUse': ['agent-comms-commander send-guard'],
         'UserPromptSubmit': ['agent-comms-watch-guard'],
-        'PostToolUse': ['agent-comms-watch-guard']}
+        'PostToolUse': ['agent-comms-watch-guard'],
+        'PreCompact': ['agent-comms-claude-precompact'],
+        'SessionStart': ['agent-comms-claude-compacted']}
     assert hooks['PreToolUse'][0]['matcher'] == 'Bash'
     assert 'matcher' not in hooks['PostToolUse'][0]
+    assert hooks['SessionStart'][0]['matcher'] == 'compact'