feat(presence): server-side last_seen stamping for non-CC agents

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 ──────────────────────────────────────

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():

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

src/pinky_daemon/codex_session.py

@@ -62,13 +62,15 @@ 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
         self._cost_callback = cost_callback
         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,
         *,

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,
         *,

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):

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"})

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: