diff --git a/pyproject.toml b/pyproject.toml index 90fe6d07c9..6058cd38fa 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/agent_comms/cli_bridges/shared.py b/src/agent_comms/cli_bridges/shared.py index e48dbc3430..19b2e3f6fb 100644 --- a/src/agent_comms/cli_bridges/shared.py +++ b/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) diff --git a/src/agent_comms/registration.py b/src/agent_comms/registration.py index 91dc1ad65a..f23ca32bbd 100644 --- a/src/agent_comms/registration.py +++ b/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. diff --git a/src/agent_comms/thread_execution.py b/src/agent_comms/thread_execution.py index 7606a790a8..1605c8b7eb 100644 --- a/src/agent_comms/thread_execution.py +++ b/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 diff --git a/src/agent_comms/thread_status.py b/src/agent_comms/thread_status.py index 3e79341069..c1dcd399d9 100644 --- a/src/agent_comms/thread_status.py +++ b/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. diff --git a/src/agent_comms_hooks/compaction.py b/src/agent_comms_hooks/compaction.py new file mode 100644 index 0000000000..a4f8ecdaf9 --- /dev/null +++ b/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()) diff --git a/src/agent_comms_hooks/watch.py b/src/agent_comms_hooks/watch.py index ef65dd78a5..102a8f0524 100644 --- a/src/agent_comms_hooks/watch.py +++ b/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,7 +57,7 @@ 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() @@ -61,6 +65,9 @@ def state(self) -> WatchState: 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) diff --git a/tests/test_thread_status.py b/tests/test_thread_status.py index 1e8bd583a6..680cde7cd7 100644 --- a/tests/test_thread_status.py +++ b/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)"]) diff --git a/tests/test_watch_guard.py b/tests/test_watch_guard.py index d95680877f..7c4d3222bb 100644 --- a/tests/test_watch_guard.py +++ b/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' diff --git a/tools/probe_claude_monitor_compaction.py b/tools/probe_claude_monitor_compaction.py new file mode 100644 index 0000000000..d15b2b5fba --- /dev/null +++ b/tools/probe_claude_monitor_compaction.py @@ -0,0 +1,145 @@ +"""Real continuous Claude Monitor -> /compact, on a new private bus/session. + +The parent logs Claude's actual streaming output; nothing injects hook payloads. +Use LABEL SOURCE. Scratch ownership is recorded in docs/bridge-compaction.md. +""" +from __future__ import annotations + +from contextlib import redirect_stdout +from io import StringIO +import json +import os +from pathlib import Path +import queue +import subprocess +import sys +import threading +import time +import argparse +from uuid import uuid4 + +PYTHON = Path('/home/ts/code/projects/agent-comms/stack/.venv/bin/python') +BIN = PYTHON.parent +SCRATCH = Path('/home/ts/.cache/agent-scratch/pi-bridge-compact') + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('label') + parser.add_argument('source', type=Path) + parser.add_argument('--compaction-hooks', action='store_true') + parser.add_argument('--claude', default='claude', help='Pin the reporting session executable when reproducing') + options = parser.parse_args() + source = options.source.resolve() + # Warm the actual CLI entrypoint once. Spawning a fresh interpreter every + # second costs more than a second on this host and distorts the observation. + sys.path.insert(0, str(source / 'src')) + from agent_comms.cli import main as cli_main + + run = SCRATCH / options.label + run.mkdir(parents=True, exist_ok=False) + root = run / 'bus' + session = str(uuid4()) + (run / 'session').write_text(session) + env = {key: value for key, value in os.environ.items() + if not key.startswith(('CLAUDE_', 'AGENT_COMMS_', 'PI_')) and key != 'CLAUDECODE'} + env.update(AGENT_COMMS_ROOT=str(root), PYTHONPATH=str(Path(source).resolve() / 'src')) + settings = run / 'settings.json' + logger = run / 'hook.py' + logger.write_text('import io,json,sys,time\nfrom pathlib import Path\n' + f'p=Path({str(run / "hooks.jsonl")!r})\n' + 'record=json.load(sys.stdin)\n' + 'with p.open("a") as f: f.write(json.dumps({"at":time.time(),"hook":record})+"\\n")\n' + + ('sys.stdin=io.StringIO(json.dumps(record))\n' + 'from agent_comms_hooks.compaction import starting,finished\n' + 'if record["hook_event_name"] == "PreCompact": starting()\n' + 'elif record.get("source") == "compact": finished()\n' + if options.compaction_hooks else '')) + settings.write_text(json.dumps({'hooks': {event: [{'hooks': [{'type': 'command', + 'command': f'{PYTHON} {logger}'}]}] for event in ('PreCompact', 'SessionStart')}})) + subprocess.run([str(BIN / 'agent-comms'), '--root', str(root), 'register', '--name', + 'compact-proof', '--worktree', str(run)], env=env, check=True, capture_output=True) + with redirect_stdout(StringIO()): + if cli_main(['--root', str(root), 'threads']): + raise RuntimeError('CLI warm-up failed') + command = [options.claude, '-p', '--input-format', 'stream-json', '--output-format', 'stream-json', + '--verbose', '--settings', str(settings), '--setting-sources', '', '--session-id', session, + '--permission-mode', 'bypassPermissions', '--strict-mcp-config', '--mcp-config', '{"mcpServers":{}}', + '--tools', 'Monitor,Bash', + '--system-prompt', 'You are a verification assistant. Follow the user literally.'] + replies = queue.Queue() + with (run / 'claude.jsonl').open('w') as log, (run / 'stderr.log').open('w') as errors: + child = subprocess.Popen(command, env=env, cwd=run, stdin=subprocess.PIPE, + stdout=subprocess.PIPE, stderr=errors, text=True, bufsize=1) + def read(): + for line in child.stdout: + log.write(line) + log.flush() + event = json.loads(line) + if event.get('type') == 'result': + replies.put(event) + replies.put(None) + reader = threading.Thread(target=read, daemon=True) + reader.start() + def send(text): + child.stdin.write(json.dumps({'type': 'user', 'message': {'role': 'user', 'content': text}}) + '\n') + child.stdin.flush() + bridge_command = (f'{BIN / "agent-comms-claude-bridge"} watch --state {run / "bridge"} ' + f'--root {root} --bus-thread compact-proof --session {session} 2>&1') + seed = ('Use the Monitor tool to run exactly this command with timeout_ms=1800000. ' + 'Do not use Bash for the watch and do not start a second watch. Then reply only WATCHING.\n' + + bridge_command + '\nKeep these constraints in context:\n' + '\n'.join( + f'{i}. Preserve original sessions, auth, uncertain attempts and messages. ' + 'Only the connected bridge owns presence; a UI displays canonical state.' for i in range(100))) + try: + send(seed) + first = replies.get(timeout=150) + print('first result:', first, flush=True) + if first is None: + raise RuntimeError('Claude exited before the watch was established') + with (run / 'statuses.jsonl').open('w') as status: + start = time.monotonic() + sent_at = finished_at = None + for second in range(120): + due = start + second - time.monotonic() + if due > 0: + time.sleep(due) + view = StringIO() + with redirect_stdout(view): + code = cli_main(['--root', str(root), 'threads']) + if code: + raise RuntimeError(view.getvalue()) + observed = json.loads(view.getvalue()) + row = {'second': second, 'at': time.time(), 'threads': observed} + status.write(json.dumps(row) + '\n') + status.flush() + print(json.dumps(row), flush=True) + # Wait for actual registration, not Claude's WATCHING text. + if sent_at is None and second >= 4 and observed['threads'][0]['status'] == 'connected': + send('/compact') + sent_at = second + if sent_at is not None and finished_at is None: + try: + result = replies.get_nowait() + except queue.Empty: + pass + else: + if result is None or result.get('local_command') == 'compact': + print('compact result:', result, flush=True) + finished_at = second + if finished_at is not None and second >= finished_at + 5: + break + else: + raise TimeoutError('Original compaction unfinished; never replay it') + finally: + child.stdin.close() + try: + child.wait(timeout=10) + except subprocess.TimeoutExpired: + child.terminate() + child.wait(timeout=10) + reader.join(timeout=5) + + +if __name__ == '__main__': + main()