diff --git a/src/pinky_daemon/agent_registry.py b/src/pinky_daemon/agent_registry.py index 9dbf4e9d..bd9e823b 100644 --- a/src/pinky_daemon/agent_registry.py +++ b/src/pinky_daemon/agent_registry.py @@ -184,6 +184,7 @@ class Agent: retired_at: float = 0.0 # When was this agent retired working_status: str = "idle" # idle, working, offline working_status_updated_at: float = 0.0 # When working_status last changed + last_seen_at: float = 0.0 # Server-side presence: updated on delivery/turn completion provider_url: str = "" # e.g. "http://localhost:11434" for Ollama, empty = Anthropic default provider_key: str = "" # API key override, empty = use ANTHROPIC_API_KEY env var provider_model: str = "" # model name override (e.g. "llama3.2"), empty = use agent.model @@ -241,6 +242,7 @@ def to_dict(self) -> dict: "retired_at": self.retired_at, "working_status": self.working_status, "working_status_updated_at": self.working_status_updated_at, + "last_seen_at": self.last_seen_at, "provider_url": self.provider_url, "provider_key": self.provider_key, "provider_model": self.provider_model, @@ -726,6 +728,7 @@ def _migrate(self) -> None: ("disallowed_tools", "TEXT NOT NULL DEFAULT '[]'"), ("thinking_effort", "TEXT NOT NULL DEFAULT 'medium'"), ("watchdog_config", "TEXT NOT NULL DEFAULT '{}'"), + ("last_seen_at", "REAL NOT NULL DEFAULT 0"), ] for col, typedef in migrations: if col not in existing: @@ -1058,7 +1061,7 @@ def register(self, name: str, **kwargs) -> Agent: "librarian_enabled, librarian_schedule, " "working_status, working_status_updated_at, " "provider_url, provider_key, provider_model, provider_ref, " - "disallowed_tools, thinking_effort, watchdog_config" + "disallowed_tools, thinking_effort, watchdog_config, last_seen_at" ) def get(self, name: str) -> Agent | None: @@ -1127,6 +1130,12 @@ def restore(self, name: str) -> bool: return True return False + def stamp_last_seen(self, name: str, ts: float | None = None) -> None: + """Server-side presence: stamp agents.last_seen_at. Agent-agnostic.""" + ts = ts if ts is not None else time.time() + self._db.execute("UPDATE agents SET last_seen_at = ? WHERE name = ?", (ts, name)) + self._db.commit() + def set_working_status(self, name: str, status: str) -> bool: """Update an agent's working status (idle, working, offline).""" if status not in ("idle", "working", "offline"): @@ -2581,6 +2590,7 @@ def _row_to_agent(self, row: tuple) -> Agent: disallowed_tools=json.loads(row[43]) if len(row) > 43 and row[43] else [], thinking_effort=row[44] if len(row) > 44 and row[44] else "medium", watchdog_config=json.loads(row[45]) if len(row) > 45 and row[45] else {}, + last_seen_at=row[46] if len(row) > 46 else 0.0, ) # ── Cost Tracking ────────────────────────────────────── diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index 27669145..edad45cd 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -2491,6 +2491,7 @@ async def _start_streaming_session( "conversation_store": store, "cost_callback": _make_cost_callback(agents), "analytics_store": analytics, + "registry": agents, } if is_codex: init_kwargs["stream_event_callback"] = await _make_streaming_event_callback(agent_name, label) @@ -5173,19 +5174,35 @@ def _agent_presence(agent_name: str) -> dict: live = broker.get_live_agents() streaming = agent_name in live hb = agents.get_latest_heartbeat(agent_name) + agent_obj = agents.get(agent_name) + hb_ts = hb.timestamp if hb else 0 + server_ts = agent_obj.last_seen_at if agent_obj else 0 + last_seen = max(server_ts, hb_ts) + now = time.time() if streaming: status = "online" - elif hb: - age = time.time() - hb.timestamp + elif hb and hb_ts >= server_ts: + # Heartbeat is the freshest liveness signal — honor its status field. + age = now - hb.timestamp if hb.status == "alive" and age < 300: status = "online" elif hb.status == "alive" and age < 1800: status = "idle" else: status = "offline" + elif server_ts > 0: + # Server-stamped activity (turn completion, delivered inter-agent msg, etc.) + # is authoritative liveness evidence for agents that don't emit heartbeats. + age = now - server_ts + if age < 300: + status = "online" + elif age < 1800: + status = "idle" + else: + status = "offline" else: status = "unknown" - return {"status": status, "streaming": streaming, "last_seen": hb.timestamp if hb else 0} + return {"status": status, "streaming": streaming, "last_seen": last_seen} @app.get("/agents/presence") async def get_all_agent_presence(): diff --git a/src/pinky_daemon/broker.py b/src/pinky_daemon/broker.py index b5eab626..94b9ae5c 100644 --- a/src/pinky_daemon/broker.py +++ b/src/pinky_daemon/broker.py @@ -808,6 +808,11 @@ async def _route_streaming(self, agent_name: str, message: BrokerMessage) -> Non message_id=message.message_id, agent_hint=hint, ) + # Server-side presence: successful inbound delivery = agent pipe is working + try: + self._registry.stamp_last_seen(agent_name) + except Exception as e: + _log(f"broker: stamp_last_seen failed for {agent_name}: {e}") # Start typing indicator for Telegram chats if message.chat_id: await self._start_typing(agent_name, message.platform, message.chat_id, streaming) @@ -828,6 +833,11 @@ async def inject_agent_message( ts = datetime.now(tz.utc).strftime("%Y-%m-%d %H:%M:%S UTC") prompt = f"[agent | {from_agent} | internal | {ts}]\n{message}" await streaming.send(prompt) + # Server-side presence: successful delivery = agent is reachable + try: + self._registry.stamp_last_seen(to_agent) + except Exception as e: + _log(f"broker: stamp_last_seen failed for {to_agent}: {e}") self._stats["routed"] += 1 _log(f"broker: injected agent message {from_agent} -> {to_agent}") return True diff --git a/src/pinky_daemon/codex_session.py b/src/pinky_daemon/codex_session.py index 4d99a1f2..ae1dc5bf 100644 --- a/src/pinky_daemon/codex_session.py +++ b/src/pinky_daemon/codex_session.py @@ -62,6 +62,7 @@ def __init__( cost_callback=None, # fn(agent_name, cost_usd, input_tokens, output_tokens, session_id) stream_event_callback=None, # async fn(event: dict) for incremental UI streaming analytics_store=None, + registry=None, # AgentRegistry — for server-side presence stamping ) -> None: self._config = config self._response_callback = response_callback @@ -69,6 +70,7 @@ def __init__( self._conversation_store = conversation_store self._stream_event_callback = stream_event_callback self._analytics_store = analytics_store + self._registry = registry self._connected = False self._processing = False # True while a codex exec is running self._message_queue: asyncio.Queue[tuple[str, str, str, str]] = asyncio.Queue() @@ -680,6 +682,7 @@ async def _handle_event(self, event: dict, result: CodexTurnResult) -> None: "cached_input_tokens": result.cached_input_tokens, }, ) + self._stamp_last_seen() await self._emit_stream_event({ "type": "turn_completed", "agent": self.agent_name, @@ -697,6 +700,7 @@ async def _handle_event(self, event: dict, result: CodexTurnResult) -> None: err_msg = err.get("message", "turn failed") result.errors.append(err_msg) self._analytics_log_activity("turn_failed", metadata={"error": err_msg}) + self._stamp_last_seen() await self._emit_stream_event({ "type": "turn_failed", "agent": self.agent_name, @@ -708,6 +712,7 @@ async def _handle_event(self, event: dict, result: CodexTurnResult) -> None: err_msg = event.get("message", "unknown error") result.errors.append(err_msg) self._analytics_log_activity("turn_error", metadata={"error": err_msg}) + self._stamp_last_seen() await self._emit_stream_event({ "type": "turn_error", "agent": self.agent_name, @@ -933,6 +938,15 @@ def _analytics_log_activity(self, event_type: str, *, metadata: dict | None = No except Exception as e: _log(f"codex[{self.agent_name}]: analytics activity failed: {e}") + def _stamp_last_seen(self) -> None: + """Server-side presence: stamp agent last_seen_at (agent-agnostic).""" + if not self._registry: + return + try: + self._registry.stamp_last_seen(self.agent_name) + except Exception as e: + _log(f"codex[{self.agent_name}]: stamp_last_seen failed: {e}") + def _analytics_log_turn_usage( self, *, diff --git a/src/pinky_daemon/streaming_session.py b/src/pinky_daemon/streaming_session.py index fd910242..d0c6f79f 100644 --- a/src/pinky_daemon/streaming_session.py +++ b/src/pinky_daemon/streaming_session.py @@ -173,12 +173,14 @@ def __init__( conversation_store=None, # ConversationStore for history logging cost_callback=None, # fn(agent_name, cost_usd, input_tokens, output_tokens, session_id) analytics_store=None, + registry=None, # AgentRegistry — for server-side presence stamping ) -> None: self._config = config self._response_callback = response_callback self._cost_callback = cost_callback # Sync callback to persist costs self._conversation_store = conversation_store self._analytics_store = analytics_store + self._registry = registry self._client = None self._reader_task: asyncio.Task | None = None self._connected = False @@ -548,6 +550,7 @@ async def _reader_loop(self) -> None: "errors": str(msg.errors or "")[:200], }, ) + self._stamp_last_seen() self._last_response = "" self._current_activity = "" self._activity_log = [] @@ -663,6 +666,8 @@ async def _reader_loop(self) -> None: "cached_input_tokens": agg_cached, }, ) + # Stamp on any turn end (complete or error) — proves pipe is live + self._stamp_last_seen() # Log assistant response to conversation store with metadata if self._last_response and self._conversation_store: @@ -960,6 +965,15 @@ def _analytics_log_activity( except Exception as e: _log(f"streaming[{self.agent_name}]: analytics activity failed: {e}") + def _stamp_last_seen(self) -> None: + """Server-side presence: stamp agent last_seen_at (agent-agnostic).""" + if not self._registry: + return + try: + self._registry.stamp_last_seen(self.agent_name) + except Exception as e: + _log(f"streaming[{self.agent_name}]: stamp_last_seen failed: {e}") + def _analytics_log_turn_usage( self, *, diff --git a/tests/test_agent_registry.py b/tests/test_agent_registry.py index 6dca4166..286d7f43 100644 --- a/tests/test_agent_registry.py +++ b/tests/test_agent_registry.py @@ -128,6 +128,28 @@ def test_to_dict(self, registry): assert d["model"] == "opus" assert d["enabled"] is True + def test_stamp_last_seen_updates_column(self, registry): + registry.register("seen") + agent = registry.get("seen") + assert agent.last_seen_at == 0.0 + + registry.stamp_last_seen("seen", ts=1234567.0) + updated = registry.get("seen") + assert updated.last_seen_at == 1234567.0 + + def test_stamp_last_seen_default_ts_is_now(self, registry): + import time as _time + registry.register("seen2") + before = _time.time() + registry.stamp_last_seen("seen2") + after = _time.time() + updated = registry.get("seen2") + assert before <= updated.last_seen_at <= after + + def test_stamp_last_seen_missing_agent_is_noop(self, registry): + # Should not raise — just affects zero rows. + registry.stamp_last_seen("ghost", ts=42.0) + class TestDirectives: def test_add_directive(self, registry): diff --git a/tests/test_api.py b/tests/test_api.py index 3efc07f7..5f6730f8 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1438,6 +1438,67 @@ def test_all_agents_presence(self): data = resp.json() assert "agents" in data + def _make_client_with_registry(self): + """Return (client, registry) both pointing at the same in-api agents.db.""" + from pinky_daemon.agent_registry import AgentRegistry + from pinky_daemon.api import create_api + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + app = create_api(max_sessions=10, default_working_dir="/tmp", db_path=path) + # create_api splits storage: agents live in {db_path}_agents.db + registry = AgentRegistry(db_path=path.replace(".db", "_agents.db")) + return TestClient(app), registry + + def test_agent_presence_server_stamped_online(self): + """Non-heartbeat agent with fresh last_seen_at should be online, not unknown.""" + client, registry = self._make_client_with_registry() + client.post("/agents", json={"name": "codex-a", "model": "opus"}) + registry.stamp_last_seen("codex-a", ts=time.time()) + resp = client.get("/agents/codex-a/presence") + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "online", f"expected online, got {data['status']}" + assert data["last_seen"] > 0 + assert data["streaming"] is False + + def test_agent_presence_server_stamped_idle(self): + """Non-heartbeat agent stamped 10min ago should be idle.""" + client, registry = self._make_client_with_registry() + client.post("/agents", json={"name": "codex-b", "model": "opus"}) + registry.stamp_last_seen("codex-b", ts=time.time() - 600) + resp = client.get("/agents/codex-b/presence") + data = resp.json() + assert data["status"] == "idle", f"expected idle, got {data['status']}" + + def test_agent_presence_server_stamped_offline(self): + """Non-heartbeat agent stamped 1hr ago should be offline.""" + client, registry = self._make_client_with_registry() + client.post("/agents", json={"name": "codex-c", "model": "opus"}) + registry.stamp_last_seen("codex-c", ts=time.time() - 3600) + resp = client.get("/agents/codex-c/presence") + data = resp.json() + assert data["status"] == "offline", f"expected offline, got {data['status']}" + + def test_agent_presence_no_stamp_no_heartbeat_is_unknown(self): + """Preserve existing behavior: no server stamp and no heartbeat → unknown.""" + client = self._make_client() + client.post("/agents", json={"name": "ghost", "model": "sonnet"}) + resp = client.get("/agents/ghost/presence") + data = resp.json() + assert data["status"] == "unknown" + + def test_agent_presence_heartbeat_wins_when_fresher(self): + """If heartbeat is fresher than server stamp, heartbeat status logic applies.""" + client, registry = self._make_client_with_registry() + client.post("/agents", json={"name": "cc-agent", "model": "sonnet"}) + # Old server stamp + registry.stamp_last_seen("cc-agent", ts=time.time() - 3600) + # Fresh heartbeat — should override server stamp + registry.record_heartbeat("cc-agent", status="alive") + resp = client.get("/agents/cc-agent/presence") + data = resp.json() + assert data["status"] == "online" + def test_agent_directives_crud(self): client = self._make_client() client.post("/agents", json={"name": "alice", "model": "sonnet"}) diff --git a/tests/test_broker.py b/tests/test_broker.py index 9f05a8ba..910ce97c 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -94,6 +94,38 @@ async def test_route_response_skips_plain_text_when_outreach_used(self): finally: tmpdir.cleanup() + @pytest.mark.asyncio + async def test_inject_agent_message_stamps_last_seen_on_success(self): + tmpdir, registry, broker, _, _ = self._make_broker() + try: + class _FakeStreaming: + is_connected = True + sent: list[str] = [] + + async def send(self, prompt: str) -> None: + _FakeStreaming.sent.append(prompt) + + broker.register_streaming("barsik", _FakeStreaming(), label="main") + assert registry.get("barsik").last_seen_at == 0.0 + + ok = await broker.inject_agent_message("pushok", "barsik", "hi") + assert ok is True + assert registry.get("barsik").last_seen_at > 0.0 + assert _FakeStreaming.sent # delivery happened + finally: + tmpdir.cleanup() + + @pytest.mark.asyncio + async def test_inject_agent_message_does_not_stamp_when_not_connected(self): + tmpdir, registry, broker, _, _ = self._make_broker() + try: + # No streaming session registered — inject should fail without stamping. + ok = await broker.inject_agent_message("pushok", "barsik", "hi") + assert ok is False + assert registry.get("barsik").last_seen_at == 0.0 + finally: + tmpdir.cleanup() + def test_remember_message_context_tracks_voice_and_reply_metadata(self): tmpdir, _, broker, _, _ = self._make_broker() try: diff --git a/tests/test_codex_session.py b/tests/test_codex_session.py index d75680f8..d89e45a6 100644 --- a/tests/test_codex_session.py +++ b/tests/test_codex_session.py @@ -164,6 +164,66 @@ async def test_error_item(self): r = await self._parse_events(events) assert "bad model" in r.errors + @pytest.mark.asyncio + async def test_error_event_stamps_last_seen(self): + """Transport/runtime errors (event_type=error) should stamp last_seen — + it's a real liveness signal even when no turn.failed fires.""" + stamps: list[str] = [] + + class _MockRegistry: + def stamp_last_seen(self, name: str, ts: float | None = None) -> None: + stamps.append(name) + + config = StreamingSessionConfig( + agent_name="test-agent", working_dir="/tmp", provider_url="codex_cli", + ) + session = CodexSession(config, registry=_MockRegistry()) + result = CodexTurnResult() + await session._handle_event( + {"type": "error", "message": "transport closed"}, result, + ) + assert stamps == ["test-agent"] + assert "transport closed" in result.errors[0] + + @pytest.mark.asyncio + async def test_turn_completed_stamps_last_seen(self): + """Regression guard: turn.completed continues to stamp.""" + stamps: list[str] = [] + + class _MockRegistry: + def stamp_last_seen(self, name: str, ts: float | None = None) -> None: + stamps.append(name) + + config = StreamingSessionConfig( + agent_name="test-agent", working_dir="/tmp", provider_url="codex_cli", + ) + session = CodexSession(config, registry=_MockRegistry()) + result = CodexTurnResult() + await session._handle_event( + {"type": "turn.completed", "usage": {"input_tokens": 5, "output_tokens": 2}}, + result, + ) + assert stamps == ["test-agent"] + + @pytest.mark.asyncio + async def test_turn_failed_stamps_last_seen(self): + """Regression guard: turn.failed continues to stamp.""" + stamps: list[str] = [] + + class _MockRegistry: + def stamp_last_seen(self, name: str, ts: float | None = None) -> None: + stamps.append(name) + + config = StreamingSessionConfig( + agent_name="test-agent", working_dir="/tmp", provider_url="codex_cli", + ) + session = CodexSession(config, registry=_MockRegistry()) + result = CodexTurnResult() + await session._handle_event( + {"type": "turn.failed", "error": {"message": "rate limited"}}, result, + ) + assert stamps == ["test-agent"] + class TestCodexSessionSendDrop: """Verify send() drops messages when not connected."""