From a7659b40c896b3387fa55934315e168de3155527 Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Mon, 20 Jul 2026 15:50:32 -0700 Subject: [PATCH 01/10] Stage 1+2: make DB the primary conversation store (Slack-off content + rebuild) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Historically Slack was the primary durable store of conversation content: the in-memory MessageLog was rebuilt from Slack history on every start, and agent_messages held only metadata (message_length, no body). With Slack off the system could not reconstruct any conversation and a restart lost all history. This lands the foundation from specs/local-db-conversations.md so the local DB is the single source of truth and the simulation can run with Slack fully off. Stage 1 — content persistence + DB rebuild: - Migration 0019 extends agent_messages with content, sender_name, is_bot, posted_at and nullable slack_ts/slack_channel_id/slack_thread_ts mirror columns; relaxes agent_id to nullable (NULL = human/PI, mirroring LogEntry.sender_agent_id); adds UNIQUE(simulation_run_id, message_ts) plus posted_at / channel / partial slack_ts indexes. - MessageLog.append is now idempotent (skips duplicate ts) and fires a persist callback; load_entry() appends restored rows without re-persisting. - SimulationEngine buffers every appended entry (bot, peer, and human) and batch-upserts them into agent_messages in _flush_persisted (ON CONFLICT on run+message_ts), drained each main-loop tick and on stop(). The per-post _log_message path is retired. - _rebuild_state_from_db() hydrates the log from agent_messages; the per-agent state reconstruction is extracted to _rebuild_agent_state() so it runs with Slack on or off; _rebuild_state_from_slack is demoted to a Slack-only reconcile that only adds messages missing from the DB. Stage 2 — local id minting: - mint_ts() yields monotonic, unique, ts-shaped ids seeded from the rebuild's max(posted_at); replaces the str(time.time()) fallbacks and removes the "mock_ts" constant (a latent collision that idempotent append would have turned into message loss). - Slack-off channels use stable local: ids (can't collide with Slack C…/G…); seeded channels are persisted to agent_channels. Verification: full suite 308 passed (new tests cover idempotent append, the persist callback, load_entry bypass, and mint_ts monotonicity/seeding), plus a real-Postgres continuity check (post -> persist -> fresh rebuild with thread intact, duplicate dropped, human row stored with NULL agent_id). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../versions/0019_agent_message_content.py | 85 +++++ specs/local-db-conversations.md | 137 ++++++++ src/agent/message_log.py | 39 ++- src/agent/simulation.py | 311 ++++++++++++++---- src/agent/slack_client.py | 10 +- src/models/agent_activity.py | 46 ++- tests/test_message_log.py | 32 ++ tests/test_private_channel_migration.py | 7 +- tests/test_simulation_logic.py | 22 ++ 9 files changed, 612 insertions(+), 77 deletions(-) create mode 100644 alembic/versions/0019_agent_message_content.py create mode 100644 specs/local-db-conversations.md diff --git a/alembic/versions/0019_agent_message_content.py b/alembic/versions/0019_agent_message_content.py new file mode 100644 index 0000000..17c52a3 --- /dev/null +++ b/alembic/versions/0019_agent_message_content.py @@ -0,0 +1,85 @@ +"""Add conversation-content columns to agent_messages (DB becomes primary store) + +Revision ID: 0019 +Revises: 0018 +Create Date: 2026-07-20 00:00:00.000000 + +Makes the local DB the primary store for agent conversations: agent_messages now +carries the message body and sender metadata (previously only in Slack + the +in-memory MessageLog), plus nullable Slack-mirror mapping columns. See +specs/local-db-conversations.md. +""" + +from typing import Sequence, Union + +import sqlalchemy as sa + +from alembic import op + +revision: str = "0019" +down_revision: Union[str, None] = "0018" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Content columns — DB is now the durable conversation store. + op.add_column( + "agent_messages", + sa.Column("content", sa.Text(), nullable=False, server_default=""), + ) + op.add_column( + "agent_messages", + sa.Column("sender_name", sa.String(100), nullable=False, server_default=""), + ) + op.add_column( + "agent_messages", + sa.Column("is_bot", sa.Boolean(), nullable=False, server_default=sa.text("true")), + ) + op.add_column( + "agent_messages", + sa.Column("posted_at", sa.Float(), nullable=False, server_default="0"), + ) + # Slack-mirror mapping (NULL when Slack is off / message is DB-origin). + op.add_column("agent_messages", sa.Column("slack_ts", sa.String(50), nullable=True)) + op.add_column("agent_messages", sa.Column("slack_channel_id", sa.String(100), nullable=True)) + op.add_column("agent_messages", sa.Column("slack_thread_ts", sa.String(50), nullable=True)) + + # agent_id becomes the sender_agent_id: NULL for human/PI messages. + op.alter_column("agent_messages", "agent_id", existing_type=sa.String(50), nullable=True) + + # Idempotency + rebuild/mirror indexes. + op.create_unique_constraint( + "uq_agent_messages_run_ts", "agent_messages", ["simulation_run_id", "message_ts"] + ) + op.create_index( + "ix_agent_messages_run_posted", + "agent_messages", + ["simulation_run_id", "posted_at"], + ) + op.create_index( + "ix_agent_messages_run_channel_posted", + "agent_messages", + ["simulation_run_id", "channel_name", "posted_at"], + ) + op.create_index( + "ix_agent_messages_run_slack_ts", + "agent_messages", + ["simulation_run_id", "slack_ts"], + postgresql_where=sa.text("slack_ts IS NOT NULL"), + ) + + +def downgrade() -> None: + op.drop_index("ix_agent_messages_run_slack_ts", table_name="agent_messages") + op.drop_index("ix_agent_messages_run_channel_posted", table_name="agent_messages") + op.drop_index("ix_agent_messages_run_posted", table_name="agent_messages") + op.drop_constraint("uq_agent_messages_run_ts", "agent_messages", type_="unique") + op.alter_column("agent_messages", "agent_id", existing_type=sa.String(50), nullable=False) + op.drop_column("agent_messages", "slack_thread_ts") + op.drop_column("agent_messages", "slack_channel_id") + op.drop_column("agent_messages", "slack_ts") + op.drop_column("agent_messages", "posted_at") + op.drop_column("agent_messages", "is_bot") + op.drop_column("agent_messages", "sender_name") + op.drop_column("agent_messages", "content") diff --git a/specs/local-db-conversations.md b/specs/local-db-conversations.md new file mode 100644 index 0000000..c368e8f --- /dev/null +++ b/specs/local-db-conversations.md @@ -0,0 +1,137 @@ +# Local-DB-Backed Conversations (Slack as an optional mirror) + +Status: in progress. Companion to `specs/agent-system.md` and +`specs/privacy-and-channel-visibility.md`. + +## Motivation + +Historically the agent simulation treated **Slack as the primary durable store +of conversation content**. The in-memory `MessageLog` (`src/agent/message_log.py`) +held message text only for the life of the process and was **rebuilt from Slack +history on every startup** (`SimulationEngine._rebuild_state_from_slack`). The +database was a metadata-only sibling: `agent_messages` stored `message_length` +but no content, and only the agent's *own* posts were recorded. With Slack off, +the system could not reconstruct any conversation and a restart lost all history. + +This spec makes the local PostgreSQL database the **single source of truth** for +conversations. The simulation must run identically with **all Slack API access +disabled**. When Slack is enabled it is a redundant, bidirectional mirror/view: + +- **Outbound:** DB-origin messages are posted to Slack for human viewing. +- **Inbound:** human/PI Slack messages are written into the DB. + +The DB reproduces the Slack-provided semantics the engine relies on — channels, +threads, membership/permissions (`public` vs `collab_private`), and PI↔bot DMs — +by **reusing the existing schema** wherever possible. + +## Core design rules + +1. **Canonical id = the id a message is born with.** A DB-origin message + (Slack-off, or an agent post in mirror mode) gets a locally-minted, ts-shaped + id from `mint_ts()`. A Slack-origin message keeps its Slack `ts` as canonical. + In pure Slack-on mode `message_ts == slack_ts`, so behavior is identical to the + pre-change system. Every structure keyed by ts (`PostRef.post_id`, + `ThreadState.thread_id`, `_poll_cursors`, `ThreadDecision.thread_id`, + `MessageLog._by_ts`) is unchanged. + +2. **`mint_ts()` is monotonic and unique.** `val = max(time.time(), + _last_mint_ts + 1e-6)`, `_last_mint_ts` seeded at rebuild from `max(posted_at)` + so minted ids sort after restored history. This preserves `posted_at = + float(ts)` ordering. + +3. **Persist at the single chokepoint `MessageLog.append`.** Peer, human/PI, and + reopen messages reach state only via `append`. A persist callback there + (mirroring `set_bot_name_map`) captures all senders and keeps `message_log.py` + DB-agnostic. DMs never enter `MessageLog`, so they persist separately. + +4. **`append` is idempotent** (skip in-memory add and callback when `ts` is + already present) — safe only because `mint_ts()` guarantees uniqueness. + +5. **Reuse, don't replace.** Extend `agent_messages` rather than add a parallel + table. The `visibility` model, `private_channel_members`, `_visibility_permits`, + thread participant rules, and `_sync_private_channels_from_db` all port directly. + +## Schema + +### `agent_messages` (extended — migration 0019) + +Adds, all NOT NULL with server defaults so existing rows survive: +`content Text ''`, `sender_name String(100) ''`, `is_bot Boolean true`, +`posted_at Float 0` (ordering key), and nullable mirror columns +`slack_ts String(50)`, `slack_channel_id String(100)`, `slack_thread_ts String(50)`. +`agent_id` is relaxed to nullable and read as `sender_agent_id` (NULL = human/PI). +`message_ts` is the canonical id. + +Indexes: `UNIQUE(simulation_run_id, message_ts)`, +`INDEX(simulation_run_id, posted_at)`, +`INDEX(simulation_run_id, channel_name, posted_at)`, +partial `INDEX(simulation_run_id, slack_ts) WHERE slack_ts IS NOT NULL`. + +### `pi_dm_messages` (new — migration 0020) + +`id`, `simulation_run_id` (FK cascade), `agent_id`, `pi_user_id` (Slack user id +Slack-on; `local:` off), `direction` enum(`inbound`,`outbound`), +`content`, `sender_name`, `ts` (canonical), `slack_ts` (nullable), `posted_at`, +`created_at`. Indexes `(simulation_run_id, agent_id, posted_at)` and +`(simulation_run_id, direction, posted_at)`. + +### Channels + +No schema change. Slack-off stores `local:` in `agent_channels.channel_id`. +Public subscriptions stay re-derived from profile keywords at seeding; a +`channel_subscriptions` table is deferred until hand-edited/agent-created public +subscriptions must survive a restart independent of re-derivation. + +## Transport abstraction + +`src/agent/transport.py` defines a `Transport` `Protocol` covering the exact +method set the engine calls on `AgentSlackClient`: + +- outbound: `post_message`, `send_dm`, `create_channel`, `create_private_channel`, + `invite_to_channel`, `join_channel`, `list_channels`, `open_dm_channel` +- inbound: `poll_channel_messages`, `get_thread_replies`, `get_all_thread_replies`, + `get_full_channel_history`, `poll_dm_messages` +- identity: `connect`, `is_connected`, `bot_user_id`, `resolve_user_name`, + `is_bot_user` + +`SlackTransport` is today's `AgentSlackClient` conformed. `NullTransport` reports +`is_connected=False` / `bot_user_id=None` (so existing +`if client and client.is_connected` branches take the no-op path), returns a +minted-id dict from outbound calls, and `[]` from inbound calls. + +`slack_enabled` (config + CLI) auto-detects from the presence of ≥1 valid token +with an explicit override; `--mock`/no-token ⇒ Slack-off. + +## PI interaction without Slack + +When Slack is off, PIs interact through a web interface that **writes inbound +rows** (`agent_messages` with `is_bot=false`/`sender_agent_id=null`, or +`pi_dm_messages` `direction='inbound'`). The engine's `_poll_pi_inbox_from_db()` +reads those new rows each tick and routes them through the existing PI-handling +logic (`_check_pi_proposal_review`, `has_pi_directive`, thread reopen, `@bot` tag +→ `handle_channel_tag`/`handle_dm`). This is the convergence point: the Slack +mirror's inbound side and the Slack-off PI path are the same DB reader. Identity +is `AgentRegistry.user_id` rather than `slack_user_id`. + +## Rollout (staged, each independently shippable) + +1. Content persistence + DB rebuild (`_rebuild_state_from_db`; demote + `_rebuild_state_from_slack` to a Slack-gated reconcile). +2. Local id minting (`mint_ts`, remove `mock_ts` constant, `local:` channel ids). +3. Transport abstraction + `slack_enabled` + `_poll_pi_inbox_from_db`. +4. Slack-less private-channel migration branch. +5. PI web interface. +6. Secondary Slack posters guarded by `slack_enabled`; outbound mirror write-back + (records `slack_ts`; reconcile dedups on `slack_ts`). +7. One-time Slack→DB backfill (`scripts/backfill_slack_history_to_db.py`). + +## Verification + +- Slack-off boot: `python -m src.agent.main --mock --fresh --max-runtime 1` + starts, runs `_rebuild_state_from_db`, seeds `local:` channels, persists content. +- Continuity: two Slack-off runs — resume rebuilds the log with content and + active threads intact. +- Parity: Slack-on produces identical DB/log state and `message_ts == slack_ts`. +- Existing suites stay green (`tests/test_message_log.py`, + `test_simulation_logic.py`, `test_private_channel_migration.py`, + `test_roster_sync.py`, `test_privacy_scoping.py`, `test_thread_not_found.py`). diff --git a/src/agent/message_log.py b/src/agent/message_log.py index 55c2d25..2d700c1 100644 --- a/src/agent/message_log.py +++ b/src/agent/message_log.py @@ -3,7 +3,7 @@ import logging import re from dataclasses import dataclass, field -from typing import Any +from typing import Any, Callable logger = logging.getLogger(__name__) @@ -45,13 +45,46 @@ def __init__(self) -> None: self._by_ts: dict[str, LogEntry] = {} # ts -> entry for fast lookup # Map bot_name (lowercase) -> agent_id, set by SimulationEngine self._bot_name_to_id: dict[str, str] = {} + # Optional persistence hook, invoked once per *new* append. The engine + # registers this to mirror the log into the DB (the primary store). + # Kept as a plain callback so this module stays DB-agnostic. See + # specs/local-db-conversations.md. + self._persist_cb: Callable[[LogEntry], None] | None = None def set_bot_name_map(self, mapping: dict[str, str]) -> None: """Register bot_name -> agent_id mapping (lowercase keys).""" self._bot_name_to_id = dict(mapping) - def append(self, entry: LogEntry) -> None: - """Add a message to the log.""" + def set_persist_callback(self, cb: Callable[[LogEntry], None] | None) -> None: + """Register a callback fired after each new append (for DB persistence).""" + self._persist_cb = cb + + def append(self, entry: LogEntry) -> bool: + """Add a message to the log. + + Idempotent: if an entry with this ts is already present, the append is + skipped (both the in-memory add and the persist callback) and False is + returned. This unifies the previously scattered ``get_entry`` guards and + keeps the DB persist hook from double-writing during Slack reconciliation. + Safe because ids are unique (Slack ts or a minted ts; see mint_ts). + Returns True when a new entry was added. + """ + if entry.ts in self._by_ts: + return False + self._entries.append(entry) + self._by_ts[entry.ts] = entry + if self._persist_cb is not None: + self._persist_cb(entry) + return True + + def load_entry(self, entry: LogEntry) -> None: + """Append a restored entry WITHOUT firing the persist callback. + + Used by the DB-rebuild path so rows just read from the DB are not + re-persisted. Still idempotent on ts. + """ + if entry.ts in self._by_ts: + return self._entries.append(entry) self._by_ts[entry.ts] = entry diff --git a/src/agent/simulation.py b/src/agent/simulation.py index dc6493d..3b40f6c 100644 --- a/src/agent/simulation.py +++ b/src/agent/simulation.py @@ -213,6 +213,14 @@ def __init__( # add/remove of agents as their status flips). See _sync_roster_from_db. self._last_roster_poll: float = 0.0 + # DB persistence buffer for the message log. MessageLog.append fires a + # sync callback that enqueues here; _flush_persisted() batch-writes to + # agent_messages once per main-loop tick. This makes the DB the primary + # conversation store. See specs/local-db-conversations.md. + self._pending_persist: list[LogEntry] = [] + # Monotonic id minter high-water mark (seeded at DB rebuild). See mint_ts. + self._last_mint_ts: float = 0.0 + # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ @@ -271,6 +279,7 @@ async def start(self) -> None: # Setup self._ensure_seeded_channels() + await self._persist_seeded_channels() # Load any collab_private channels created via the web-UI reopen flow # BEFORE rebuilding state so the rebuild's history-fetch loop covers # them too — otherwise the handover message wouldn't land in the @@ -278,7 +287,14 @@ async def start(self) -> None: await self._sync_private_channels_from_db() self._build_lab_directories() await self._load_pi_mappings() + # The DB is the primary conversation store. Register the persist hook, + # hydrate the log from the DB, then (only when Slack is connected) + # reconcile with Slack history, and finally reconstruct per-agent state + # from the combined log. This whole sequence runs with Slack fully off. + self.message_log.set_persist_callback(self._enqueue_persist) + await self._rebuild_state_from_db() await self._rebuild_state_from_slack() + await self._rebuild_agent_state() # Rebuild advanced last_seen_cursor to max(all_messages), which can # overshoot messages in private channels (typically older than the # latest public chatter). Rewind member-bot cursors so Phase 2 can @@ -387,7 +403,8 @@ async def start(self) -> None: elif settings.turn_delay_seconds > 0: await asyncio.sleep(settings.turn_delay_seconds) - # Flush LLM logs periodically + # Flush buffered message-log entries + LLM logs periodically + await self._flush_persisted() if self._llm_log_buffer: await self._flush_llm_logs() @@ -397,6 +414,7 @@ async def stop(self) -> None: """Stop the simulation gracefully.""" self._running = False set_call_log_callback(None) + await self._flush_persisted() await self._flush_llm_logs() logger.info("Simulation stopping...") @@ -2249,6 +2267,20 @@ async def _poll_proposal_threads_for_pi(self) -> None: # Message posting # ------------------------------------------------------------------ + def mint_ts(self) -> str: + """Return a monotonic, unique, ts-shaped id (decimal seconds string). + + The canonical message/channel id when there is no Slack ts (Slack-off, + or a DB-origin message). Monotonicity preserves the posted_at=float(ts) + ordering the engine relies on; _last_mint_ts is seeded from the rebuild's + max(posted_at) so new ids always sort after restored history. Uniqueness + is what makes the idempotent MessageLog.append safe. + See specs/local-db-conversations.md. + """ + val = max(time.time(), self._last_mint_ts + 1e-6) + self._last_mint_ts = val + return f"{val:.6f}" + async def _post_message( self, agent_id: str, @@ -2281,7 +2313,15 @@ async def _post_message( else: logger.info("[%s] MOCK post to #%s: %s...", agent_id, channel, text[:60]) - ts = result.get("ts", str(time.time())) if result else str(time.time()) + # Canonical id: the Slack ts when a connected client posted, else a + # locally-minted ts. Slack ts (when present) is also recorded as the + # mirror mapping on the entry. + slack_ts = result.get("ts") if result else None + ts = slack_ts or self.mint_ts() + try: + posted_at = float(ts) + except (TypeError, ValueError): + posted_at = time.time() # Add to message log entry = LogEntry( @@ -2291,23 +2331,13 @@ async def _post_message( sender_name=agent.bot_name if agent else f"{agent_id}Bot", content=text, thread_ts=thread_ts, - posted_at=float(ts) if ts else time.time(), + posted_at=posted_at, is_bot=True, ) + # Persisted to agent_messages via the MessageLog append callback + # (_enqueue_persist → _flush_persisted). The DB is the primary store. self.message_log.append(entry) - # Log to database - if self.session_factory and self.simulation_run_id: - await self._log_message( - agent_id=agent_id, - channel_id=result.get("channel", channel) if result else channel, - channel_name=channel, - message_ts=ts, - thread_ts=thread_ts, - message_length=len(text), - phase="thread_reply" if thread_ts else "new_post", - ) - # ------------------------------------------------------------------ # Setup helpers # ------------------------------------------------------------------ @@ -2349,8 +2379,9 @@ def _ensure_seeded_channels(self) -> None: """Create any missing seeded channels and join relevant bots.""" client = next(iter(self.slack_clients.values()), None) if not client or not client.is_connected: - # Mock mode — populate channel map with fake IDs - self._channel_id_map = {ch: f"mock_{ch}" for ch in SEEDED_CHANNELS} + # Slack off — channels are DB-native with stable local: ids that + # can't collide with Slack C…/G… ids. See specs/local-db-conversations.md. + self._channel_id_map = {ch: f"local:{ch}" for ch in SEEDED_CHANNELS} # All seeded channels are public. self._channel_visibility = {ch: VISIBILITY_PUBLIC for ch in SEEDED_CHANNELS} return @@ -2381,6 +2412,45 @@ def _ensure_seeded_channels(self) -> None: for c in self.slack_clients.values(): c._channel_name_to_id.update(existing) + async def _persist_seeded_channels(self) -> None: + """Record seeded channels in agent_channels for this run (idempotent). + + Keeps channel existence in the DB so the workspace is reconstructable + without Slack (and so the admin UI can count channels). Uses the current + _channel_id_map (Slack ids when on, local: ids when off). + """ + if not self.session_factory or not self.simulation_run_id: + return + from sqlalchemy import select as sa_select + from src.agent.channels import record_channel_created + try: + async with self.session_factory() as db: + existing_names = set( + (await db.execute( + sa_select(AgentChannel.channel_name).where( + AgentChannel.simulation_run_id == self.simulation_run_id + ) + )).scalars().all() + ) + created = 0 + for ch_name in SEEDED_CHANNELS: + if ch_name in existing_names: + continue + await record_channel_created( + db, + simulation_run_id=self.simulation_run_id, + channel_id=self._channel_id_map.get(ch_name, f"local:{ch_name}"), + channel_name=ch_name, + channel_type="thematic", + created_by_agent="system", + ) + created += 1 + if created: + await db.commit() + logger.info("Persisted %d seeded channels to agent_channels", created) + except Exception as exc: + logger.warning("Failed to persist seeded channels: %s", exc) + def _build_lab_directories(self) -> None: """Build a condensed publications directory for each agent (excluding their own lab).""" lab_pubs: dict[str, list[str]] = {} @@ -2431,11 +2501,155 @@ async def _backfill_foa_cache(self) -> None: except Exception as exc: logger.warning("FOA cache backfill failed: %s", exc) + async def _rebuild_state_from_db(self) -> None: + """Hydrate the MessageLog from agent_messages — the primary store. + + Loads message bodies (available since migration 0019) via the + callback-bypassing path so restored rows aren't re-persisted. Seeds the + mint_ts high-water mark and, for rows that were mirrored to Slack, the + Slack poll cursors so a later Slack reconcile only fetches newer messages. + See specs/local-db-conversations.md. + """ + if not self.session_factory or not self.simulation_run_id: + logger.info("No DB session — skipping DB rebuild") + return + from sqlalchemy import select as sa_select + try: + async with self.session_factory() as db: + result = await db.execute( + sa_select(AgentMessage) + .where(AgentMessage.simulation_run_id == self.simulation_run_id) + .order_by(AgentMessage.posted_at.asc(), AgentMessage.created_at.asc()) + ) + rows = result.scalars().all() + except Exception as exc: + logger.warning("DB rebuild failed: %s", exc) + return + + loaded = 0 + max_posted = 0.0 + for r in rows: + # Pre-0019 rows carry only metadata (empty body): skip them. They + # hold no conversational signal, and if Slack is on the reconcile + # pass re-adds them with content. Stage 7 backfills legacy content. + if not r.content or not r.message_ts: + continue + entry = LogEntry( + ts=r.message_ts, + channel=r.channel_name, + sender_agent_id=r.agent_id, + sender_name=r.sender_name or "", + content=r.content, + thread_ts=r.thread_ts, + posted_at=r.posted_at or 0.0, + is_bot=r.is_bot, + visibility=r.visibility, + ) + self.message_log.load_entry(entry) + loaded += 1 + if entry.posted_at > max_posted: + max_posted = entry.posted_at + # Advance the Slack poll cursor for rows that were mirrored, so the + # optional reconcile only fetches genuinely newer Slack messages. + if r.slack_ts and r.slack_channel_id: + cur = self._poll_cursors.get(r.slack_channel_id, "0") + if r.slack_ts > cur: + self._poll_cursors[r.slack_channel_id] = r.slack_ts + self._last_mint_ts = max(self._last_mint_ts, max_posted) + logger.info("Rebuilt MessageLog from DB: %d messages", loaded) + + async def _flush_persisted(self) -> None: + """Batch-upsert buffered message-log entries into agent_messages. + + Uses ON CONFLICT (simulation_run_id, message_ts) so it is safe to run + alongside legacy rows, transitional double-writes, and repeated restarts. + Drops the buffer when there is no DB so it can't grow unbounded. + """ + if not self._pending_persist: + return + if not self.session_factory or not self.simulation_run_id: + self._pending_persist.clear() + return + entries = self._pending_persist + self._pending_persist = [] + # Dedup by canonical id within the batch — a single ON CONFLICT statement + # cannot touch the same row twice. + by_ts: dict[str, dict] = {} + for e in entries: + if not e.ts: + continue + channel_id = self._channel_id_map.get(e.channel) or f"local:{e.channel}" + by_ts[e.ts] = { + "simulation_run_id": self.simulation_run_id, + "agent_id": e.sender_agent_id, + "channel_id": channel_id, + "channel_name": e.channel, + "message_ts": e.ts, + "message_length": len(e.content or ""), + "thread_ts": e.thread_ts, + "phase": "thread_reply" if e.thread_ts else "new_post", + "visibility": e.visibility, + "content": e.content or "", + "sender_name": e.sender_name or "", + "is_bot": e.is_bot, + "posted_at": e.posted_at, + } + rows = list(by_ts.values()) + if not rows: + return + from sqlalchemy import func as sa_func + from sqlalchemy import select as sa_select + from sqlalchemy.dialects.postgresql import insert as pg_insert + try: + async with self.session_factory() as db: + stmt = pg_insert(AgentMessage.__table__).values(rows) + stmt = stmt.on_conflict_do_update( + constraint="uq_agent_messages_run_ts", + set_={ + "content": stmt.excluded.content, + "sender_name": stmt.excluded.sender_name, + "is_bot": stmt.excluded.is_bot, + "posted_at": stmt.excluded.posted_at, + "message_length": stmt.excluded.message_length, + "visibility": stmt.excluded.visibility, + "thread_ts": stmt.excluded.thread_ts, + "channel_id": stmt.excluded.channel_id, + "channel_name": stmt.excluded.channel_name, + "agent_id": stmt.excluded.agent_id, + }, + ) + await db.execute(stmt) + # Keep the run's message total accurate (bulk upsert can't easily + # distinguish inserts from updates, so recompute the count). + run = (await db.execute( + sa_select(SimulationRun).where(SimulationRun.id == self.simulation_run_id) + )).scalar_one_or_none() + if run: + total = (await db.execute( + sa_select(sa_func.count(AgentMessage.id)).where( + AgentMessage.simulation_run_id == self.simulation_run_id + ) + )).scalar_one() + run.total_messages = total + run.total_api_calls = sum(a.api_call_count for a in self.agents.values()) + await db.commit() + except Exception as exc: + logger.warning("Failed to flush %d messages: %s", len(rows), exc) + + def _enqueue_persist(self, entry: LogEntry) -> None: + """MessageLog persist callback — buffer a new entry for the next flush.""" + self._pending_persist.append(entry) + async def _rebuild_state_from_slack(self) -> None: - """Rebuild MessageLog and agent state from Slack history + DB.""" + """Reconcile the MessageLog with Slack history (Slack-on only). + + The DB is the primary store (_rebuild_state_from_db); this pass only + adds messages that exist on Slack but not yet in the log — via the + idempotent append, which also persists them to the DB. + """ default_client = next(iter(self.slack_clients.values()), None) if not default_client or not default_client.is_connected: - logger.info("No Slack client available — skipping state rebuild") + logger.info("No Slack client available — skipping Slack reconcile") return # Build a mapping of bot_user_id -> agent_id @@ -2524,11 +2738,18 @@ async def _rebuild_state_from_slack(self) -> None: total_messages += 1 logger.info( - "Rebuilt MessageLog: %d messages across %d channels, %d threads", + "Slack reconcile: appended %d messages across %d channels, %d threads", total_messages, len(polled_ids), total_threads, ) - # 2. Rebuild active_threads per agent + async def _rebuild_agent_state(self) -> None: + """Reconstruct per-agent state from the message log + DB. + + Runs after both the DB rebuild and the optional Slack reconcile, so it + behaves identically with Slack on or off. Reads only self.message_log, + thread_decisions, proposal_reviews and llm_call_logs — no Slack calls. + """ + # Rebuild active_threads per agent. # Get all closed thread IDs and prior thread summaries from thread_decisions closed_thread_ids: set[str] = set() if self.session_factory: @@ -3021,14 +3242,15 @@ async def _sync_proposal_reviews_from_db(self) -> None: # Create a synthetic log entry for the PI guidance so it appears # in thread history and the agents can see it + minted = self.mint_ts() pi_entry = LogEntry( - ts=str(time.time()), + ts=minted, channel=channel, sender_agent_id=None, sender_name="PI (via web)", content=guidance, thread_ts=thread_id, - posted_at=time.time(), + posted_at=float(minted), is_bot=False, ) self.message_log.append(pi_entry) @@ -3171,49 +3393,6 @@ def _seed_private_refinements(self, migrated_info: dict[str, tuple[str, str]]) - # poster (the counterpart will be seeded once they're loaded). self._db_private_refined_thread_ids.add(thread_id) - async def _log_message( - self, - agent_id: str, - channel_id: str, - channel_name: str, - message_ts: str | None, - thread_ts: str | None, - message_length: int, - phase: str, - ) -> None: - """Log an agent message to the database.""" - if not self.session_factory or not self.simulation_run_id: - return - try: - async with self.session_factory() as db: - record = AgentMessage( - simulation_run_id=self.simulation_run_id, - agent_id=agent_id, - channel_id=channel_id, - channel_name=channel_name, - message_ts=message_ts, - thread_ts=thread_ts, - message_length=message_length, - phase=phase, - ) - db.add(record) - # Update run totals - from sqlalchemy import select - run_result = await db.execute( - select(SimulationRun).where( - SimulationRun.id == self.simulation_run_id - ) - ) - run = run_result.scalar_one_or_none() - if run: - run.total_messages = (run.total_messages or 0) + 1 - run.total_api_calls = sum( - a.api_call_count for a in self.agents.values() - ) - await db.commit() - except Exception as exc: - logger.warning("Failed to log message: %s", exc) - # ------------------------------------------------------------------ # Post-simulation # ------------------------------------------------------------------ diff --git a/src/agent/slack_client.py b/src/agent/slack_client.py index e1779e2..8dfc928 100644 --- a/src/agent/slack_client.py +++ b/src/agent/slack_client.py @@ -367,8 +367,12 @@ def post_message( ) -> dict | None: """Post a message to a Slack channel (accepts name or ID).""" if not self._client: + # Not connected: report "not posted" so the engine mints a unique + # canonical id via mint_ts (a hardcoded ts here would collide and, + # under idempotent append, drop real messages). See + # specs/local-db-conversations.md. logger.info("[%s] MOCK post to #%s: %s", self.agent_id, channel, text[:80]) - return {"ts": "mock_ts", "channel": channel} + return None channel_id = self._resolve_channel_id(channel) # Ensure bot is in the channel. Skipped for private channels, which @@ -466,7 +470,7 @@ def create_channel(self, name: str) -> dict | None: """Create a new Slack channel.""" if not self._client: logger.info("[%s] MOCK create channel: #%s", self.agent_id, name) - return {"id": f"mock_{name}", "name": name} + return {"id": f"local:{name}", "name": name} try: result = self._client.conversations_create(name=name) ch = result["channel"] @@ -500,7 +504,7 @@ def create_private_channel(self, name: str) -> dict | None: candidate = f"{name[: 80 - len(suffix)].rstrip('-')}{suffix}" if not self._client: logger.info("[%s] MOCK create private channel: #%s", self.agent_id, candidate) - return {"id": f"mock_priv_{candidate}", "name": candidate, "is_private": True} + return {"id": f"local:{candidate}", "name": candidate, "is_private": True} try: result = self._call_with_retry( self._client.conversations_create, name=candidate, is_private=True, diff --git a/src/models/agent_activity.py b/src/models/agent_activity.py index 79fbb82..8f773d4 100644 --- a/src/models/agent_activity.py +++ b/src/models/agent_activity.py @@ -3,7 +3,21 @@ import uuid from datetime import datetime -from sqlalchemy import CheckConstraint, DateTime, Enum, Float, ForeignKey, Index, Integer, String, Text, func +from sqlalchemy import ( + Boolean, + CheckConstraint, + DateTime, + Enum, + Float, + ForeignKey, + Index, + Integer, + String, + Text, + UniqueConstraint, + func, + text, +) from sqlalchemy.dialects.postgresql import JSON, UUID from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -62,9 +76,14 @@ class AgentMessage(Base): ForeignKey("simulation_runs.id", ondelete="CASCADE"), nullable=False, ) - agent_id: Mapped[str] = mapped_column(String(50), nullable=False) + # Nullable: the sender's agent_id, or NULL for human/PI messages + # (mirrors LogEntry.sender_agent_id). Every reader filters for a specific + # agent_id, so NULL rows are naturally excluded. See specs/local-db-conversations.md. + agent_id: Mapped[str | None] = mapped_column(String(50), nullable=True) channel_id: Mapped[str] = mapped_column(String(100), nullable=False) channel_name: Mapped[str] = mapped_column(String(100), nullable=False) + # Canonical message id: a locally-minted ts-shaped string (Slack-off) or the + # Slack ts (Slack-on). Unique within a run. message_ts: Mapped[str | None] = mapped_column(String(50), nullable=True) message_length: Mapped[int] = mapped_column(Integer, default=0, nullable=False) thread_ts: Mapped[str | None] = mapped_column(String(50), nullable=True) @@ -72,10 +91,33 @@ class AgentMessage(Base): visibility: Mapped[str] = mapped_column( String(20), nullable=False, default=VISIBILITY_PUBLIC, ) # denormalized from agent_channels.visibility; see specs/privacy-and-channel-visibility.md §G1/G2 + # Content columns (DB is now the primary conversation store, not Slack). + content: Mapped[str] = mapped_column(Text, nullable=False, server_default="") + sender_name: Mapped[str] = mapped_column(String(100), nullable=False, server_default="") + is_bot: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default="true") + posted_at: Mapped[float] = mapped_column(Float, nullable=False, server_default="0") + # Slack mirror mapping (NULL when Slack is off / message is DB-origin). + slack_ts: Mapped[str | None] = mapped_column(String(50), nullable=True) + slack_channel_id: Mapped[str | None] = mapped_column(String(100), nullable=True) + slack_thread_ts: Mapped[str | None] = mapped_column(String(50), nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False ) + __table_args__ = ( + UniqueConstraint("simulation_run_id", "message_ts", name="uq_agent_messages_run_ts"), + Index("ix_agent_messages_run_posted", "simulation_run_id", "posted_at"), + Index( + "ix_agent_messages_run_channel_posted", + "simulation_run_id", "channel_name", "posted_at", + ), + Index( + "ix_agent_messages_run_slack_ts", + "simulation_run_id", "slack_ts", + postgresql_where=text("slack_ts IS NOT NULL"), + ), + ) + # Relationships simulation_run: Mapped["SimulationRun"] = relationship( "SimulationRun", back_populates="messages" diff --git a/tests/test_message_log.py b/tests/test_message_log.py index 2e055c4..d061f26 100644 --- a/tests/test_message_log.py +++ b/tests/test_message_log.py @@ -208,3 +208,35 @@ def test_skips_human_messages(self, log): ) log.append(human) assert log.get_last_bot_sender_in_channel("priv-x") == "su" + + +# --------------------------------------------------------------- +# Idempotent append + persist callback (DB-primary store) +# --------------------------------------------------------------- + +class TestAppendIdempotencyAndPersist: + def test_append_returns_true_then_false_on_duplicate_ts(self, log): + assert log.append(_post("1", "general", "su", "SuBot", "first")) is True + # Same ts: skipped, returns False, no duplicate stored. + assert log.append(_post("1", "general", "su", "SuBot", "dup")) is False + assert len(log) == 1 + # The original content is retained (the duplicate is dropped). + assert log.get_entry("1").content == "first" + + def test_persist_callback_fires_once_per_new_append(self, log): + seen = [] + log.set_persist_callback(lambda e: seen.append(e.ts)) + log.append(_post("1", "general", "su", "SuBot", "a")) + log.append(_post("1", "general", "su", "SuBot", "a-dup")) # skipped + log.append(_post("2", "general", "wiseman", "WisemanBot", "b")) + assert seen == ["1", "2"] + + def test_load_entry_bypasses_callback(self, log): + seen = [] + log.set_persist_callback(lambda e: seen.append(e.ts)) + log.load_entry(_post("1", "general", "su", "SuBot", "restored")) + assert len(log) == 1 + assert seen == [] # rebuild path must not re-persist + # Still idempotent on ts. + log.load_entry(_post("1", "general", "su", "SuBot", "again")) + assert len(log) == 1 diff --git a/tests/test_private_channel_migration.py b/tests/test_private_channel_migration.py index 3621709..e59ea39 100644 --- a/tests/test_private_channel_migration.py +++ b/tests/test_private_channel_migration.py @@ -209,15 +209,16 @@ def test_returns_mock_channel_with_is_private(self, mock_client): # Mock mode applies the same timestamp suffix as the live path. assert ch["name"].startswith("priv-test-") assert ch["is_private"] is True - assert ch["id"].startswith("mock_priv_") + # Slack-off channels use the DB-native 'local:' id scheme. + assert ch["id"].startswith("local:") def test_public_create_channel_still_works(self, mock_client): """Don't regress the existing create_channel behavior.""" ch = mock_client.create_channel("general") assert ch is not None assert ch["name"] == "general" - # Mock public channels use the 'mock_' prefix (no 'priv_'). - assert ch["id"] == "mock_general" + # Slack-off channels use the DB-native 'local:' id scheme. + assert ch["id"] == "local:general" class _FakeSlack: diff --git a/tests/test_simulation_logic.py b/tests/test_simulation_logic.py index db04132..cb0bb56 100644 --- a/tests/test_simulation_logic.py +++ b/tests/test_simulation_logic.py @@ -645,3 +645,25 @@ async def test_handover_memo_is_not_treated_as_revised_proposal(self): assert self.NAME in engine._finalized_private_channels props = [p for p in lairson.state.pending_proposals if p.thread_id == "200.000002"] assert len(props) == 1 + + +# --------------------------------------------------------------- +# mint_ts — monotonic, unique, ts-shaped ids (DB-primary store) +# --------------------------------------------------------------- + +class TestMintTs: + def test_monotonic_and_unique_under_tight_loop(self): + engine = SimulationEngine(agents=[], slack_clients={}) + ids = [engine.mint_ts() for _ in range(1000)] + floats = [float(x) for x in ids] + # Strictly increasing (so posted_at=float(ts) ordering is preserved) + assert all(b > a for a, b in zip(floats, floats[1:])) + # All unique + assert len(set(ids)) == len(ids) + + def test_seeded_high_water_mark_sorts_after_history(self): + engine = SimulationEngine(agents=[], slack_clients={}) + # Simulate a rebuild that saw a far-future max(posted_at). + engine._last_mint_ts = 9_999_999_999.0 + first = float(engine.mint_ts()) + assert first > 9_999_999_999.0 From 131104fcaa8849e6077a6aed190850c9888d05f8 Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Mon, 20 Jul 2026 15:57:55 -0700 Subject: [PATCH 02/10] Stage 3: transport abstraction + slack_enabled flag + DB inbox poller MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Formalizes running with Slack fully off and adds the Slack-independent inbound path for PI messages. - New src/agent/transport.py: a Transport Protocol matching the exact surface the engine uses on AgentSlackClient (which conforms structurally, no change), plus NullTransport — a no-op used when Slack is disabled. NullTransport reports is_connected=False (so the engine's existing guards take the no-op path), returns None from outbound posts (engine mints a local id), and returns local: ids for channel creates and [] from all inbound polls. - config.slack_enabled (bool | None): None auto-detects (Slack on iff any agent has a usable token); SLACK_ENABLED=false forces DB-only mode. main.py resolves it (--mock forces off), builds AgentSlackClients when on and NullTransports when off, and passes the flag to the engine. - SimulationEngine.slack_enabled gates the roster hot-add: when off, newly-active agents are admitted with a NullTransport instead of being dropped for a missing token / failed connect. - New _poll_pi_inbox_from_db(): ingests human/PI rows the web app writes to agent_messages (is_bot=false), appends any unseen ones to the MessageLog, and routes them through PI handling (proposal-review clear, thread reopen, pi_context, @bot tags) derived from the thread's own participants rather than a Slack user→agent map. Runs every tick regardless of Slack, and is the convergence point for the future Slack mirror's inbound side. Cursor seeded past restored history in _rebuild_state_from_db. Verification: full suite 314 passed (new tests: NullTransport behavior + Protocol conformance for both NullTransport and AgentSlackClient), plus a real-Postgres smoke test (a web-written PI reply on a proposal thread is ingested and clears the pending-proposal block; re-poll is a no-op). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/agent/main.py | 35 ++++++++-- src/agent/simulation.py | 137 ++++++++++++++++++++++++++++++++++---- src/agent/transport.py | 142 ++++++++++++++++++++++++++++++++++++++++ src/config.py | 6 ++ tests/test_transport.py | 46 +++++++++++++ 5 files changed, 347 insertions(+), 19 deletions(-) create mode 100644 src/agent/transport.py create mode 100644 tests/test_transport.py diff --git a/src/agent/main.py b/src/agent/main.py index 1c34ab4..98d219a 100644 --- a/src/agent/main.py +++ b/src/agent/main.py @@ -91,16 +91,31 @@ async def _run_simulation( len(agents), "all statuses" if all_agents else "status='active'", ) - # Set up Slack clients (Web API only, no Socket Mode). Tokens come from the - # AgentRegistry row, falling back to the legacy .env/config mapping. + # Resolve whether Slack is enabled. --mock forces it off; an explicit + # SLACK_ENABLED env setting wins next; otherwise auto-detect from whether + # any agent has a usable bot token. When off, the DB is the sole store and + # no Slack API calls are made. See specs/local-db-conversations.md. + from src.services.slack_tokens import env_token, is_valid_token + + def _token_for(agent_id: str) -> str | None: + tok = roster_tokens.get(agent_id) + return tok if is_valid_token(tok) else env_token(agent_id) + + if mock: + slack_enabled = False + elif settings.slack_enabled is not None: + slack_enabled = settings.slack_enabled + else: + slack_enabled = any(is_valid_token(_token_for(a.agent_id)) for a in agents) + + # Set up transports. When Slack is on, each agent gets a Web-API client + # (Web API only, no Socket Mode); when off, a NullTransport that no-ops all + # Slack calls so the engine runs identically against the DB. slack_clients = {} - if not mock: + if slack_enabled: from src.agent.slack_client import AgentSlackClient - from src.services.slack_tokens import env_token, is_valid_token for agent in agents: - bot_token = roster_tokens.get(agent.agent_id) - if not is_valid_token(bot_token): - bot_token = env_token(agent.agent_id) + bot_token = _token_for(agent.agent_id) if is_valid_token(bot_token): client = AgentSlackClient( agent_id=agent.agent_id, @@ -112,6 +127,11 @@ async def _run_simulation( logger.warning("[%s] Slack connection failed — skipping", agent.agent_id) else: logger.warning("[%s] No valid Slack token — skipping", agent.agent_id) + else: + from src.agent.transport import NullTransport + for agent in agents: + slack_clients[agent.agent_id] = NullTransport(agent_id=agent.agent_id) + logger.info("Slack disabled — running DB-only (NullTransport for %d agents)", len(agents)) # Set up database session factory session_factory = None @@ -195,6 +215,7 @@ async def _run_simulation( session_factory=session_factory, simulation_run_id=simulation_run_id, reset_cursors=reset_cursors, + slack_enabled=slack_enabled, ) # Handle shutdown signals diff --git a/src/agent/simulation.py b/src/agent/simulation.py index 3b40f6c..2ccd6f4 100644 --- a/src/agent/simulation.py +++ b/src/agent/simulation.py @@ -128,6 +128,7 @@ def __init__( session_factory=None, simulation_run_id: uuid.UUID | None = None, reset_cursors: bool = False, + slack_enabled: bool = True, ): self.agents = {a.agent_id: a for a in agents} self.slack_clients = slack_clients @@ -136,6 +137,10 @@ def __init__( self.session_factory = session_factory self.simulation_run_id = simulation_run_id self._reset_cursors = reset_cursors + # When False, the local DB is the sole conversation store and no Slack + # API calls are made (transports are NullTransport). Drives the roster + # gate and the DB inbox poller. See specs/local-db-conversations.md. + self.slack_enabled = slack_enabled self._start_time: datetime | None = None self._running = False @@ -220,6 +225,10 @@ def __init__( self._pending_persist: list[LogEntry] = [] # Monotonic id minter high-water mark (seeded at DB rebuild). See mint_ts. self._last_mint_ts: float = 0.0 + # High-water mark (posted_at) for the DB inbox poller — the Slack- + # independent path by which human/PI messages written by the web app + # enter the simulation. See _poll_pi_inbox_from_db. + self._pi_inbox_cursor: float = 0.0 # ------------------------------------------------------------------ # Lifecycle @@ -319,11 +328,16 @@ async def start(self) -> None: turn_count = 0 consecutive_idle = 0 while self._running and self.is_within_time_limit: - # Poll Slack for PI messages (channels, DMs, and proposal threads) + # Poll Slack for PI messages (channels, DMs, and proposal threads). + # No-ops when Slack is off (NullTransport / no connected clients). await self._poll_slack_for_pi_messages() await self._poll_pi_dms() await self._poll_proposal_threads_for_pi() + # DB-native inbound path: human/PI messages written by the web app. + # Runs regardless of Slack, and is how PIs interact when Slack is off. + await self._poll_pi_inbox_from_db() + # Sync proposal reviews and any newly-created private channels from # the web app. Both are DB-driven, so a single tick picks them up. await self._sync_proposal_reviews_from_db() @@ -2046,6 +2060,96 @@ async def _poll_slack_for_pi_messages(self) -> None: except Exception as exc: logger.debug("Polling error for #%s: %s", ch_name, exc) + async def _poll_pi_inbox_from_db(self) -> None: + """Ingest human/PI messages written to the DB by the web app. + + This is the Slack-independent inbound path (and the convergence point + for the Slack mirror's inbound side): the PI web interface inserts rows + into agent_messages with is_bot=false, and this poller appends any it + hasn't seen to the MessageLog and routes them through the same PI + handling the Slack poller uses — proposal-review clearing, thread + reopen, pi_context, and @bot tags. Runs whether or not Slack is enabled. + See specs/local-db-conversations.md. + """ + if not self.session_factory or not self.simulation_run_id: + return + from sqlalchemy import select as sa_select + try: + async with self.session_factory() as db: + rows = (await db.execute( + sa_select(AgentMessage) + .where( + AgentMessage.simulation_run_id == self.simulation_run_id, + AgentMessage.is_bot.is_(False), + AgentMessage.posted_at > self._pi_inbox_cursor, + ) + .order_by(AgentMessage.posted_at.asc()) + )).scalars().all() + except Exception as exc: + logger.debug("PI inbox poll failed: %s", exc) + return + + for r in rows: + if r.posted_at > self._pi_inbox_cursor: + self._pi_inbox_cursor = r.posted_at + if not r.message_ts or self.message_log.get_entry(r.message_ts): + # Already known (e.g. the engine itself appended it) — skip + # re-processing, but the cursor has still advanced past it. + continue + entry = LogEntry( + ts=r.message_ts, + channel=r.channel_name, + sender_agent_id=None, + sender_name=r.sender_name or "PI", + content=r.content or "", + thread_ts=r.thread_ts, + posted_at=r.posted_at or 0.0, + is_bot=False, + visibility=r.visibility, + ) + self.message_log.append(entry) + logger.info( + "PI (web) message in #%s: %.60s", entry.channel, entry.content[:60] + ) + await self._handle_pi_inbound_entry(entry) + + async def _handle_pi_inbound_entry(self, entry: LogEntry) -> None: + """Apply PI-message side effects, derived from the thread (no Slack map). + + Clears pending-proposal blocks, reopens closed threads, sets pi_context + on active threads, and honors @bot tags — using the thread's own + participants rather than a Slack user→agent mapping, so it works with + Slack off. + """ + # Clears any pending proposal on this thread (keyed purely by thread id). + self._check_pi_proposal_review(entry) + + thread_ts = entry.thread_ts + if thread_ts: + # Reopen a closed thread for its participants. + if thread_ts in self._closed_thread_ids: + history = self.message_log.get_thread_history(thread_ts) + participants = [ + h.sender_agent_id for h in history + if h.sender_agent_id and h.sender_agent_id in self.agents + ] + if participants: + await self._reopen_thread(participants[0], thread_ts, entry) + else: + # Active thread → treat the PI message as authoritative context. + for agent in self.agents.values(): + thread = agent.state.active_threads.get(thread_ts) + if thread: + thread.pi_context = entry.content + thread.has_pending_reply = True + agent.state.has_pi_directive = True + + # @bot tag → route to the tagged agent (same as the Slack path). + tagged_id = self.message_log._extract_tagged_agent(entry.content) + if tagged_id and tagged_id in self.agents and self._pi_handler: + self.agents[tagged_id].state.has_pi_directive = True + await self._pi_handler.handle_channel_tag(tagged_id, entry) + def _check_pi_proposal_review(self, entry: LogEntry) -> None: """Check if a PI message clears a pending proposal for any agent.""" thread_ts = entry.thread_ts @@ -2556,6 +2660,9 @@ async def _rebuild_state_from_db(self) -> None: if r.slack_ts > cur: self._poll_cursors[r.slack_channel_id] = r.slack_ts self._last_mint_ts = max(self._last_mint_ts, max_posted) + # Start the inbox poller past all restored history so it only picks up + # genuinely new web-written PI messages. + self._pi_inbox_cursor = max(self._pi_inbox_cursor, max_posted) logger.info("Rebuilt MessageLog from DB: %d messages", loaded) async def _flush_persisted(self) -> None: @@ -3086,17 +3193,23 @@ async def _sync_roster_from_db(self) -> None: # --- Additions: agent newly active ------------------------------ for aid in to_add: r = desired[aid] - token = r.slack_bot_token if is_valid_token(r.slack_bot_token) else env_token(aid) - if not is_valid_token(token): - logger.info( - "[roster] Agent %s is active but has no usable token yet — " - "skipping (will retry next sync once a token is set)", aid, - ) - continue - client = AgentSlackClient(agent_id=aid, bot_token=token) - if not client.connect(): - logger.warning("[roster] Slack connect failed for new agent %s — skipping", aid) - continue + if self.slack_enabled: + token = r.slack_bot_token if is_valid_token(r.slack_bot_token) else env_token(aid) + if not is_valid_token(token): + logger.info( + "[roster] Agent %s is active but has no usable token yet — " + "skipping (will retry next sync once a token is set)", aid, + ) + continue + client = AgentSlackClient(agent_id=aid, bot_token=token) + if not client.connect(): + logger.warning("[roster] Slack connect failed for new agent %s — skipping", aid) + continue + else: + # Slack off: admit the agent with a no-op transport (never + # gate on a token/connection that doesn't apply in DB-only mode). + from src.agent.transport import NullTransport + client = NullTransport(agent_id=aid) agent = Agent(agent_id=aid, bot_name=r.bot_name, pi_name=r.pi_name) # In-place inserts (PIHandler shares these dicts by reference). self.agents[aid] = agent diff --git a/src/agent/transport.py b/src/agent/transport.py new file mode 100644 index 0000000..9b25759 --- /dev/null +++ b/src/agent/transport.py @@ -0,0 +1,142 @@ +"""Message transport abstraction — decouples the engine from Slack. + +The simulation talks to a ``Transport`` rather than to Slack directly. Two +implementations exist: + +- ``SlackTransport`` — the real Slack Web API client (``AgentSlackClient`` in + ``slack_client.py`` already conforms to this Protocol structurally; no + subclassing is required). +- ``NullTransport`` — a no-op used when Slack is disabled. Outbound calls do + nothing (the engine mints a local canonical id via ``mint_ts``); inbound + polls return nothing (human/PI input arrives through the DB inbox instead). + +This lets the whole 5-phase loop, PI polling and private-channel flows run with +Slack fully off. See specs/local-db-conversations.md. +""" + +from __future__ import annotations + +import logging +from typing import Any, Callable, Protocol, runtime_checkable + +logger = logging.getLogger(__name__) + + +@runtime_checkable +class Transport(Protocol): + """The Slack surface the engine actually uses. + + Method names match ``AgentSlackClient`` exactly so it conforms without + changes and the engine's ``slack_clients`` dict needs no renaming. + """ + + agent_id: str + + # Identity / lifecycle + def connect(self) -> bool: ... + @property + def is_connected(self) -> bool: ... + @property + def bot_user_id(self) -> str | None: ... + def resolve_user_name(self, user_id: str) -> str: ... + def is_bot_user(self, user_id: str) -> bool: ... + + # Outbound + def post_message(self, channel: str, text: str, thread_ts: str | None = None) -> dict | None: ... + def send_dm(self, user_id: str, text: str) -> dict | None: ... + def open_dm_channel(self, user_id: str) -> str | None: ... + def create_channel(self, name: str) -> dict | None: ... + def create_private_channel(self, name: str) -> dict | None: ... + def invite_to_channel(self, channel_id: str, user_ids: list[str]) -> bool: ... + def join_channel(self, channel_id: str) -> None: ... + def list_channels(self, include_private: bool = False) -> dict[str, str]: ... + def get_channel_id(self, channel_name: str) -> str | None: ... + + # Inbound + def poll_channel_messages(self, channel_id: str, oldest: str = "0", limit: int = 100) -> list[dict[str, Any]]: ... + def get_thread_replies(self, channel_id: str, thread_ts: str, oldest: str = "0") -> list[dict[str, Any]]: ... + def get_full_channel_history(self, channel_id: str) -> list[dict[str, Any]]: ... + def get_all_thread_replies(self, channel_id: str, thread_ts: str) -> list[dict[str, Any]]: ... + def poll_dm_messages(self, user_id: str, oldest: str = "0", limit: int = 20) -> list[dict[str, Any]]: ... + + +class NullTransport: + """No-op transport used when Slack is disabled (DB is the sole store). + + Reports ``is_connected == False`` so the engine's existing + ``if client and client.is_connected`` branches take the no-op path, and the + Slack pollers (which filter on connected clients) simply find nothing. + Outbound posts return None so ``_post_message`` mints a local canonical id; + channel-create calls return ``local:`` ids so DB-native channels still work. + """ + + def __init__(self, agent_id: str): + self.agent_id = agent_id + # Present for parity with AgentSlackClient — the engine updates this + # shared name->id cache in _ensure_seeded_channels / sync paths. + self._channel_name_to_id: dict[str, str] = {} + + # Identity / lifecycle + def connect(self) -> bool: + return True + + @property + def is_connected(self) -> bool: + return False + + @property + def bot_user_id(self) -> str | None: + return None + + def resolve_user_name(self, user_id: str) -> str: + return user_id + + def is_bot_user(self, user_id: str) -> bool: + return False + + def set_visibility_lookup(self, lookup: Callable[[str], str | None]) -> None: + return None + + # Outbound — no external side effects + def post_message(self, channel: str, text: str, thread_ts: str | None = None) -> dict | None: + return None + + def send_dm(self, user_id: str, text: str) -> dict | None: + return None + + def open_dm_channel(self, user_id: str) -> str | None: + return None + + def create_channel(self, name: str) -> dict | None: + return {"id": f"local:{name}", "name": name} + + def create_private_channel(self, name: str) -> dict | None: + return {"id": f"local:{name}", "name": name, "is_private": True} + + def invite_to_channel(self, channel_id: str, user_ids: list[str]) -> bool: + return True + + def join_channel(self, channel_id: str) -> None: + return None + + def list_channels(self, include_private: bool = False) -> dict[str, str]: + return dict(self._channel_name_to_id) + + def get_channel_id(self, channel_name: str) -> str | None: + return self._channel_name_to_id.get(channel_name) + + # Inbound — nothing arrives via Slack; PI input comes from the DB inbox + def poll_channel_messages(self, channel_id: str, oldest: str = "0", limit: int = 100) -> list[dict[str, Any]]: + return [] + + def get_thread_replies(self, channel_id: str, thread_ts: str, oldest: str = "0") -> list[dict[str, Any]]: + return [] + + def get_full_channel_history(self, channel_id: str) -> list[dict[str, Any]]: + return [] + + def get_all_thread_replies(self, channel_id: str, thread_ts: str) -> list[dict[str, Any]]: + return [] + + def poll_dm_messages(self, user_id: str, oldest: str = "0", limit: int = 20) -> list[dict[str, Any]]: + return [] diff --git a/src/config.py b/src/config.py index a067fed..67dc775 100644 --- a/src/config.py +++ b/src/config.py @@ -60,6 +60,12 @@ class Settings(BaseSettings): slack_config_token: str = "" slack_config_refresh_token: str = "" + # Master switch for all Slack integration. None = auto-detect (Slack is on + # iff at least one agent has a usable bot token); set SLACK_ENABLED=false to + # force the DB-only mode where the local database is the sole conversation + # store and no Slack API calls are made. See specs/local-db-conversations.md. + slack_enabled: bool | None = None + # AWS SES aws_region: str = "us-east-2" ses_sender_email: str = "noreply@copi.science" diff --git a/tests/test_transport.py b/tests/test_transport.py new file mode 100644 index 0000000..f9a7a6c --- /dev/null +++ b/tests/test_transport.py @@ -0,0 +1,46 @@ +"""Tests for the message transport abstraction (Slack-off mode).""" + +from src.agent.transport import NullTransport, Transport + + +class TestNullTransport: + def test_conforms_to_protocol(self): + t = NullTransport("su") + assert isinstance(t, Transport) + + def test_reports_disconnected(self): + t = NullTransport("su") + # is_connected=False makes the engine's guards take the no-op path. + assert t.is_connected is False + assert t.bot_user_id is None + assert t.connect() is True # usable, just not Slack-backed + + def test_outbound_posts_are_noops(self): + t = NullTransport("su") + # No Slack ts — the engine mints a local canonical id instead. + assert t.post_message("general", "hi") is None + assert t.send_dm("U1", "hi") is None + assert t.open_dm_channel("U1") is None + + def test_channel_creates_use_local_ids(self): + t = NullTransport("su") + assert t.create_channel("general") == {"id": "local:general", "name": "general"} + priv = t.create_private_channel("priv-a-b") + assert priv["id"] == "local:priv-a-b" + assert priv["is_private"] is True + assert t.invite_to_channel("local:x", ["U1", "U2"]) is True + assert t.join_channel("local:x") is None + + def test_inbound_polls_return_empty(self): + t = NullTransport("su") + assert t.poll_channel_messages("local:general") == [] + assert t.get_thread_replies("local:general", "1.0") == [] + assert t.get_full_channel_history("local:general") == [] + assert t.get_all_thread_replies("local:general", "1.0") == [] + assert t.poll_dm_messages("U1") == [] + + def test_slack_client_conforms_to_protocol(self): + # The real client must structurally satisfy the same Protocol. + from src.agent.slack_client import AgentSlackClient + client = AgentSlackClient(agent_id="su", bot_token="xoxb-test") + assert isinstance(client, Transport) From 5b0c953524b293cf0e0eba15722910e09f3412c6 Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Mon, 20 Jul 2026 16:04:02 -0700 Subject: [PATCH 03/10] Stage 4: Slack-less private-channel migration + generalize DB inbound poller MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The public→collab_private migration (PI reopen flow) now works with Slack off, and a running sim ingests migration handover posts through the DB. - private_channels.py: _slack_enabled_for_migration() resolves the mode (explicit SLACK_ENABLED wins; else auto-detect from whether both bots have usable tokens). When off, _migrate_offline() creates the collab_private AgentChannel with a local: id and the same PrivateChannelMember rows as the Slack path, but makes no Slack calls: the handover posts (authored by the creator bot) and the neutral ⏸️ origin-thread close marker are written straight to agent_messages (visibility=collab_private for the handover) and refined_in_channel is set. The other PI is not DM'd (no transport). - Generalized the engine's DB inbound poller (renamed _poll_pi_inbox_from_db -> _poll_inbound_from_db): it now ingests ANY unseen message for the run, not just human rows — so bot-authored handover posts written by the web process reach the live MessageLog. Human/PI rows still additionally go through PI handling; the sim's own messages are already in the log and are skipped. Verification: full suite 314 passed, plus a real-Postgres smoke test with SLACK_ENABLED=false (offline migration creates a local: private channel with 3 members, writes 3 handover posts + origin close marker + refined_in_channel, and a running sim ingests all 3 handover posts from the DB). Also confirmed the auto-detect path still uses Slack when tokens are present. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/agent/simulation.py | 53 +++++++------- src/services/private_channels.py | 118 +++++++++++++++++++++++++++++++ 2 files changed, 145 insertions(+), 26 deletions(-) diff --git a/src/agent/simulation.py b/src/agent/simulation.py index 2ccd6f4..0c9a8bc 100644 --- a/src/agent/simulation.py +++ b/src/agent/simulation.py @@ -225,9 +225,10 @@ def __init__( self._pending_persist: list[LogEntry] = [] # Monotonic id minter high-water mark (seeded at DB rebuild). See mint_ts. self._last_mint_ts: float = 0.0 - # High-water mark (posted_at) for the DB inbox poller — the Slack- - # independent path by which human/PI messages written by the web app - # enter the simulation. See _poll_pi_inbox_from_db. + # High-water mark (posted_at) for the DB inbound poller — the Slack- + # independent path by which messages written by other processes (PI web + # interface, private-channel handover) enter the simulation. See + # _poll_inbound_from_db. self._pi_inbox_cursor: float = 0.0 # ------------------------------------------------------------------ @@ -334,9 +335,10 @@ async def start(self) -> None: await self._poll_pi_dms() await self._poll_proposal_threads_for_pi() - # DB-native inbound path: human/PI messages written by the web app. - # Runs regardless of Slack, and is how PIs interact when Slack is off. - await self._poll_pi_inbox_from_db() + # DB-native inbound path: messages written by other processes (PI + # web interface, private-channel handover). Runs regardless of Slack, + # and is how PIs interact when Slack is off. + await self._poll_inbound_from_db() # Sync proposal reviews and any newly-created private channels from # the web app. Both are DB-driven, so a single tick picks them up. @@ -2060,16 +2062,15 @@ async def _poll_slack_for_pi_messages(self) -> None: except Exception as exc: logger.debug("Polling error for #%s: %s", ch_name, exc) - async def _poll_pi_inbox_from_db(self) -> None: - """Ingest human/PI messages written to the DB by the web app. + async def _poll_inbound_from_db(self) -> None: + """Ingest messages written to the DB by other processes. - This is the Slack-independent inbound path (and the convergence point - for the Slack mirror's inbound side): the PI web interface inserts rows - into agent_messages with is_bot=false, and this poller appends any it - hasn't seen to the MessageLog and routes them through the same PI - handling the Slack poller uses — proposal-review clearing, thread - reopen, pi_context, and @bot tags. Runs whether or not Slack is enabled. - See specs/local-db-conversations.md. + The DB is the primary store, so any message this process hasn't seen — + PI messages and bot-authored handover posts written by the web app, and + (later) the Slack mirror's inbound side — must be pulled into the live + MessageLog. Human/PI messages are additionally routed through PI handling + (proposal-review clearing, thread reopen, pi_context, @bot tags). Runs + every tick regardless of Slack. See specs/local-db-conversations.md. """ if not self.session_factory or not self.simulation_run_id: return @@ -2080,38 +2081,38 @@ async def _poll_pi_inbox_from_db(self) -> None: sa_select(AgentMessage) .where( AgentMessage.simulation_run_id == self.simulation_run_id, - AgentMessage.is_bot.is_(False), AgentMessage.posted_at > self._pi_inbox_cursor, ) .order_by(AgentMessage.posted_at.asc()) )).scalars().all() except Exception as exc: - logger.debug("PI inbox poll failed: %s", exc) + logger.debug("Inbound DB poll failed: %s", exc) return for r in rows: if r.posted_at > self._pi_inbox_cursor: self._pi_inbox_cursor = r.posted_at if not r.message_ts or self.message_log.get_entry(r.message_ts): - # Already known (e.g. the engine itself appended it) — skip - # re-processing, but the cursor has still advanced past it. + # Already known (the engine itself appended and flushed it) — + # skip re-processing, but the cursor has still advanced past it. continue entry = LogEntry( ts=r.message_ts, channel=r.channel_name, - sender_agent_id=None, - sender_name=r.sender_name or "PI", + sender_agent_id=r.agent_id, + sender_name=r.sender_name or ("PI" if not r.is_bot else r.agent_id or "bot"), content=r.content or "", thread_ts=r.thread_ts, posted_at=r.posted_at or 0.0, - is_bot=False, + is_bot=r.is_bot, visibility=r.visibility, ) self.message_log.append(entry) - logger.info( - "PI (web) message in #%s: %.60s", entry.channel, entry.content[:60] - ) - await self._handle_pi_inbound_entry(entry) + if r.is_bot: + logger.info("External bot message in #%s: %.60s", entry.channel, entry.content[:60]) + else: + logger.info("PI (web) message in #%s: %.60s", entry.channel, entry.content[:60]) + await self._handle_pi_inbound_entry(entry) async def _handle_pi_inbound_entry(self, entry: LogEntry) -> None: """Apply PI-message side effects, derived from the thread (no Slack map). diff --git a/src/services/private_channels.py b/src/services/private_channels.py index 76f8bd3..27a177f 100644 --- a/src/services/private_channels.py +++ b/src/services/private_channels.py @@ -29,6 +29,7 @@ from __future__ import annotations import logging +import time import uuid from dataclasses import dataclass @@ -40,6 +41,7 @@ from src.config import get_settings from src.models import ( AgentChannel, + AgentMessage, AgentRegistry, PrivateChannelMember, SimulationRun, @@ -215,6 +217,108 @@ async def _resolve_other_pi( return reg, user +async def _slack_enabled_for_migration( + db: AsyncSession, creator_agent_id: str, other_agent_id: str, +) -> bool: + """Resolve whether the migration should use Slack. + + Explicit SLACK_ENABLED wins; otherwise auto-detect: Slack is used only when + both participating bots have usable tokens. See specs/local-db-conversations.md. + """ + settings = get_settings() + if settings.slack_enabled is not None: + return settings.slack_enabled + from src.services.slack_tokens import get_agent_bot_token + creator_tok = await get_agent_bot_token(db, creator_agent_id) + other_tok = await get_agent_bot_token(db, other_agent_id) + return bool(creator_tok and other_tok) + + +async def _migrate_offline( + db: AsyncSession, + *, + thread_decision: ThreadDecision, + creator_agent_id: str, + creator_pi_user: User, + guidance_text: str, + a: str, + b: str, + other_agent_id: str, + origin_channel_name: str, +) -> MigrationResult: + """Slack-off migration: DB-only, no Slack calls. + + Creates the collab_private AgentChannel and members exactly as the Slack + path, but with a local: channel id, and writes the handover posts and the + origin-thread ⏸️ close marker as agent_messages rows so the running sim (and + the next rebuild) pick them up through _poll_inbound_from_db. + """ + base_slug = _build_slug(a, b, origin_channel_name) + stamp = time.strftime("%Y%m%d-%H%M%S", time.gmtime()) + new_channel_name = normalize_channel_name(f"{base_slug[: 80 - len(stamp) - 1]}-{stamp}") + new_channel_id = f"local:{new_channel_name}" + origin_channel_id = f"local:{origin_channel_name}" + + simulation_run_id = await _latest_simulation_run_id(db) + + ac = AgentChannel( + simulation_run_id=simulation_run_id, + channel_id=new_channel_id, + channel_name=new_channel_name, + channel_type="collaboration", + visibility=VISIBILITY_COLLAB_PRIVATE, + created_by_agent=creator_agent_id, + migrated_from_channel_id=origin_channel_id, + ) + db.add(ac) + await db.flush() + + db.add(PrivateChannelMember(agent_channel_id=ac.id, agent_id=creator_agent_id, role="bot")) + db.add(PrivateChannelMember(agent_channel_id=ac.id, agent_id=other_agent_id, role="bot")) + db.add(PrivateChannelMember( + agent_channel_id=ac.id, user_id=creator_pi_user.id, role="pi", + added_by_user_id=creator_pi_user.id, + )) + + # Handover posts, authored by the creator bot, written straight to the DB + # (visibility=collab_private) so they stay within the channel's membership. + handover_posts = _build_handover_messages( + creator_pi_name=creator_pi_user.name, + proposal_summary=thread_decision.summary_text, + guidance_text=guidance_text, + origin_channel_name=origin_channel_name, + ) + now = time.time() + for i, post in enumerate(handover_posts): + ts = f"{now + i * 1e-6:.6f}" + db.add(AgentMessage( + simulation_run_id=simulation_run_id, agent_id=creator_agent_id, + channel_id=new_channel_id, channel_name=new_channel_name, + message_ts=ts, phase="new_post", visibility=VISIBILITY_COLLAB_PRIVATE, + content=post, sender_name=f"{creator_agent_id}Bot", is_bot=True, + posted_at=float(ts), + )) + # Neutral close marker in the origin (public) thread — no PI text echoed. + close_ts = f"{now + len(handover_posts) * 1e-6:.6f}" + db.add(AgentMessage( + simulation_run_id=simulation_run_id, agent_id=creator_agent_id, + channel_id=origin_channel_id, channel_name=origin_channel_name, + message_ts=close_ts, thread_ts=thread_decision.thread_id, + phase="thread_reply", visibility="public", + content="⏸️ continuing this discussion off-channel.", + sender_name=f"{creator_agent_id}Bot", is_bot=True, posted_at=float(close_ts), + )) + + thread_decision.refined_in_channel = new_channel_id + logger.info("Slack-off migration: created private channel %s (DB-only)", new_channel_name) + return MigrationResult( + channel_id=new_channel_id, + channel_name=new_channel_name, + agent_channel_id=ac.id, + invited_other_pi=False, + ) + + async def migrate_public_thread_to_private( db: AsyncSession, *, @@ -243,6 +347,20 @@ async def migrate_public_thread_to_private( origin_channel_name = thread_decision.channel + # Slack-off: DB-only migration (no channel/invite/post/DM). The handover is + # written straight to agent_messages for the sim to ingest. + if not await _slack_enabled_for_migration(db, creator_agent_id, other_agent_id): + return await _migrate_offline( + db, + thread_decision=thread_decision, + creator_agent_id=creator_agent_id, + creator_pi_user=creator_pi_user, + guidance_text=guidance_text, + a=a, b=b, + other_agent_id=other_agent_id, + origin_channel_name=origin_channel_name, + ) + # --- Slack side-effects ------------------------------------------------ creator_token = await _get_or_fail_bot_token(db, creator_agent_id) other_token = await _get_or_fail_bot_token(db, other_agent_id) From cd072b9f3d20f4f9348cdb22291cb2c0963fb1ad Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Mon, 20 Jul 2026 16:09:07 -0700 Subject: [PATCH 04/10] Stage 5: PI web interface for messaging agents without Slack MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gives PIs (and delegates) a Slack-independent way to see recent activity and inject a message into their agent's workspace. - src/services/pi_inbox.py: get_latest_run_id() and record_pi_message(), which insert a human/PI row (is_bot=false, agent_id=null) into agent_messages with a minted ts, resolving channel_id/visibility from agent_channels. The running sim ingests it via _poll_inbound_from_db and routes it through PI handling. - agent_page.py: GET /{agent_id}/conversations (a read view of recent messages — content now lives in the DB — plus a post form) and POST /{agent_id}/message (writes the inbound row; an optional "address my bot" toggle prepends the @BotName tag the engine already understands). Both gated by get_agent_with_access (owner or delegate). - templates/agent/conversations.html + a dashboard card linking to it. Identity uses AgentRegistry.user_id / access checks rather than a Slack user id, so this works with Slack fully off. DM-style directives depend on DM persistence and are wired in Stage 7. Verification: full suite 314 passed; a real-Postgres smoke test (record_pi_message -> _poll_inbound_from_db ingests the PI row and the @bot tag sets has_pi_directive on the target agent); app restarted cleanly and the new routes resolve. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/routers/agent_page.py | 114 +++++++++++++++++++++++++++++ src/services/pi_inbox.py | 78 ++++++++++++++++++++ templates/agent/conversations.html | 71 ++++++++++++++++++ templates/agent/dashboard.html | 7 ++ 4 files changed, 270 insertions(+) create mode 100644 src/services/pi_inbox.py create mode 100644 templates/agent/conversations.html diff --git a/src/routers/agent_page.py b/src/routers/agent_page.py index e42e2bf..60a3fe0 100644 --- a/src/routers/agent_page.py +++ b/src/routers/agent_page.py @@ -653,6 +653,120 @@ async def reopen_proposal( # -------------------------------------------------------------------------- +@router.get("/{agent_id}/conversations", response_class=HTMLResponse) +async def agent_conversations( + agent_id: str, + request: Request, + db: AsyncSession = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Read view of the agent's recent conversations + a form to post a message. + + This is the Slack-independent way for a PI to see what their agent is + discussing and to inject a message/tag — it writes to the DB inbox, which + the running simulation ingests. See specs/local-db-conversations.md. + """ + from src.services.pi_inbox import get_latest_run_id + + agent, is_owner = await get_agent_with_access(agent_id, db, current_user) + if agent.status not in ("active", "inactive"): + return RedirectResponse(url="/agent", status_code=302) + aid = agent.agent_id + + run_id = await get_latest_run_id(db) + channels: list[str] = [] + messages: list[dict] = [] + if run_id: + # Channels this agent participates in (has authored a message in). + ch_rows = await db.execute( + select(distinct(AgentMessage.channel_name)).where( + AgentMessage.simulation_run_id == run_id, + AgentMessage.agent_id == aid, + ) + ) + channels = sorted({r[0] for r in ch_rows} | {"general"}) + # Recent messages in those channels (content is now stored in the DB). + msg_rows = await db.execute( + select(AgentMessage) + .where( + AgentMessage.simulation_run_id == run_id, + AgentMessage.channel_name.in_(channels), + ) + .order_by(AgentMessage.posted_at.desc()) + .limit(100) + ) + messages = [ + { + "channel": m.channel_name, + "sender": m.sender_name or (m.agent_id or "PI"), + "is_bot": m.is_bot, + "content": m.content, + "thread_ts": m.thread_ts, + "posted_at": m.posted_at, + } + for m in reversed(msg_rows.scalars().all()) + ] + else: + channels = ["general"] + + return templates.TemplateResponse( + request, + "agent/conversations.html", + _template_context( + request, current_user, agent=agent, is_owner=is_owner, + channels=channels, messages=messages, has_run=run_id is not None, + posted=request.query_params.get("posted"), + ), + ) + + +@router.post("/{agent_id}/message") +async def post_agent_message( + agent_id: str, + request: Request, + channel_name: str = Form(...), + content: str = Form(...), + thread_ts: str = Form(""), + tag_bot: str = Form(""), + db: AsyncSession = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Write a PI-authored message into the DB inbox for the agent's workspace. + + Ingested by the running simulation via _poll_inbound_from_db — the + Slack-independent equivalent of a PI posting in a Slack channel. + """ + from src.services.pi_inbox import get_latest_run_id, record_pi_message + + agent, is_owner = await get_agent_with_access(agent_id, db, current_user) + if agent.status != "active": + raise HTTPException(status_code=403, detail="Agent is not active") + + text = content.strip() + if not text: + raise HTTPException(status_code=400, detail="Message cannot be empty") + # Optionally address the PI's own bot so it engages (same @BotName convention + # the Slack path uses; the engine's tag detection is identical). + if tag_bot and f"@{agent.bot_name.lower()}" not in text.lower(): + text = f"@{agent.bot_name} {text}" + + run_id = await get_latest_run_id(db) + if not run_id: + raise HTTPException(status_code=409, detail="No simulation run to post into yet") + + await record_pi_message( + db, + run_id=run_id, + channel_name=channel_name.strip() or "general", + content=text, + sender_name=f"{current_user.name} (PI)", + thread_ts=thread_ts.strip() or None, + ) + await db.commit() + logger.info("[%s] PI %s posted a web message to #%s", agent_id, current_user.name, channel_name) + return RedirectResponse(url=f"/agent/{agent_id}/conversations?posted=1", status_code=302) + + @router.get("/{agent_id}/profile", response_class=HTMLResponse) async def view_private_profile( agent_id: str, diff --git a/src/services/pi_inbox.py b/src/services/pi_inbox.py new file mode 100644 index 0000000..8f48fb3 --- /dev/null +++ b/src/services/pi_inbox.py @@ -0,0 +1,78 @@ +"""Write PI-authored messages into the DB inbox (Slack-independent input path). + +The agent simulation ingests these rows via SimulationEngine._poll_inbound_from_db, +so a PI can drive their agent with Slack fully off. This is the DB-native +equivalent of the Slack channel-message path. See specs/local-db-conversations.md. +""" + +from __future__ import annotations + +import time +import uuid + +from sqlalchemy import desc, select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.models import AgentChannel, AgentMessage, SimulationRun + + +async def get_latest_run_id(db: AsyncSession) -> uuid.UUID | None: + """Return the most recent SimulationRun id, or None if there are no runs.""" + return (await db.execute( + select(SimulationRun.id).order_by(desc(SimulationRun.started_at)).limit(1) + )).scalar_one_or_none() + + +async def _resolve_channel(db: AsyncSession, run_id: uuid.UUID, channel_name: str) -> tuple[str, str]: + """Return (channel_id, visibility) for a channel name in a run. + + Falls back to a local: id / public visibility when the channel has no + agent_channels row yet (e.g. a seeded channel not persisted on an old run). + """ + row = (await db.execute( + select(AgentChannel.channel_id, AgentChannel.visibility) + .where( + AgentChannel.simulation_run_id == run_id, + AgentChannel.channel_name == channel_name, + ) + .limit(1) + )).first() + if row: + return row[0], row[1] + return f"local:{channel_name}", "public" + + +async def record_pi_message( + db: AsyncSession, + *, + run_id: uuid.UUID, + channel_name: str, + content: str, + sender_name: str, + thread_ts: str | None = None, +) -> AgentMessage: + """Insert a human/PI message (is_bot=False) into agent_messages. + + The engine's inbound poller picks it up on its next tick, appends it to the + live MessageLog, and routes it through PI handling (proposal-review clear, + thread reopen, pi_context, @bot tags). Does not commit — the caller owns the + transaction. + """ + channel_id, visibility = await _resolve_channel(db, run_id, channel_name) + ts = f"{time.time():.6f}" + msg = AgentMessage( + simulation_run_id=run_id, + agent_id=None, # human/PI sender + channel_id=channel_id, + channel_name=channel_name, + message_ts=ts, + thread_ts=thread_ts, + phase="thread_reply" if thread_ts else "new_post", + visibility=visibility, + content=content, + sender_name=sender_name, + is_bot=False, + posted_at=float(ts), + ) + db.add(msg) + return msg diff --git a/templates/agent/conversations.html b/templates/agent/conversations.html new file mode 100644 index 0000000..e1af39d --- /dev/null +++ b/templates/agent/conversations.html @@ -0,0 +1,71 @@ +{% extends "base.html" %} +{% block title %}Conversations — {{ agent.bot_name }} — CoPI{% endblock %} + +{% block content %} +
+
+
+

{{ agent.bot_name }} — Conversations

+

Post a message into your agent's workspace. Your agent picks it up on its next turn.

+
+ ← Dashboard +
+ + {% if posted %} +
+ Message posted. Your agent will see it on its next turn. +
+ {% endif %} + + {% if not has_run %} +
+ No simulation run exists yet — there's nowhere to post. Once a run starts you can message your agent here. +
+ {% endif %} + + +
+
+ + +
+ + +
+ +
+
+ + +

Recent activity

+ {% if messages %} +
+ {% for m in messages %} +
+
+ {{ m.sender }}{% if not m.is_bot %} · PI{% endif %} + #{{ m.channel }}{% if m.thread_ts %} · thread{% endif %} +
+
{{ m.content }}
+
+ {% endfor %} +
+ {% else %} +

No messages yet in your agent's channels.

+ {% endif %} +
+{% endblock %} diff --git a/templates/agent/dashboard.html b/templates/agent/dashboard.html index aebde37..512f884 100644 --- a/templates/agent/dashboard.html +++ b/templates/agent/dashboard.html @@ -292,6 +292,13 @@

Reviewed Proposals

View and edit your agent's private behavioral profile.

+ + +
Conversations
+

See recent activity and post a message to your agent — no Slack required.

+ + From c8500cd84523e399b0eadcd06ffe6d76076900ae Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Mon, 20 Jul 2026 16:15:14 -0700 Subject: [PATCH 05/10] Stage 6: guard secondary Slack posters + record Slack-mirror mapping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Makes the non-engine Slack writers degrade to the DB when Slack is off, and records the Slack↔DB id mapping so mirroring and reconcile dedup are correct. Secondary posters (gated by slack_tokens.slack_globally_enabled — explicit SLACK_ENABLED wins, else auto-detect from token presence): - grantbot.py: when Slack is off, funding opportunities are written to agent_messages (authored by a "grantbot" identity, :moneybag: top-level post) via _post_funding_to_db instead of released — so funding threads still exist and agents scan them. - email_inbound.py + agent_page.py reopen: the legacy "post guidance to the origin thread" fallback now writes to the DB inbox via record_pi_message when Slack is off (the primary reopen path already routes through the Slack-aware migration). invite.py's users_lookupByEmail was already token-guarded. Slack-mirror mapping: - LogEntry gains slack_ts/slack_channel_id; _post_message records them when a connected client posted (in pure Slack-on mode slack_ts == message_ts), and _flush_persisted upserts slack_ts/slack_channel_id/slack_thread_ts. - _rebuild_state_from_db seeds _known_slack_ts; the Slack reconcile skips any message already represented in the DB by its slack_ts (dedup for a DB-origin message later mirrored to Slack) and stamps reconciled entries with their slack mapping. Verification: full suite 314 passed, plus a real-Postgres smoke test (a connected post persists message_ts == slack_ts + slack_channel_id, and a fresh rebuild seeds _known_slack_ts for reconcile dedup). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/agent/grantbot.py | 65 ++++++++++++++++++++++----- src/agent/message_log.py | 5 +++ src/agent/simulation.py | 54 ++++++++++++++++++----- src/routers/agent_page.py | 82 ++++++++++++++++++++--------------- src/services/email_inbound.py | 20 ++++++++- src/services/slack_tokens.py | 14 ++++++ 6 files changed, 181 insertions(+), 59 deletions(-) diff --git a/src/agent/grantbot.py b/src/agent/grantbot.py index 93f41ff..27f8f2f 100644 --- a/src/agent/grantbot.py +++ b/src/agent/grantbot.py @@ -156,6 +156,30 @@ def _has_sufficient_lead_time(close_date_raw: str, now: datetime, min_days: int) return cd >= now + timedelta(days=min_days) +async def _post_funding_to_db(session: AsyncSession, channel_name: str, full_post: str) -> None: + """Write a GrantBot funding post to agent_messages (Slack-off path). + + Authored by the 'grantbot' identity as a top-level post so agents scan it in + Phase 2 and can start funding threads (funding threads are open to all). + """ + import time as _time + + from src.models import AgentMessage + from src.services.pi_inbox import get_latest_run_id + + run_id = await get_latest_run_id(session) + if not run_id: + raise RuntimeError("No simulation run to post funding opportunity into") + ts = f"{_time.time():.6f}" + session.add(AgentMessage( + simulation_run_id=run_id, agent_id="grantbot", + channel_id=f"local:{channel_name}", channel_name=channel_name, + message_ts=ts, phase="new_post", visibility="public", + content=full_post, sender_name="GrantBot", is_bot=True, posted_at=float(ts), + )) + await session.flush() + + async def _load_posted_numbers(session: AsyncSession) -> set[str]: """Return the set of already-posted FOA numbers from Postgres.""" result = await session.execute(select(GrantbotPostedFoa.foa_number)) @@ -497,19 +521,25 @@ async def _run_grantbot_with_session( logger.info("Drafted %d posts, posting %d (max %d per channel)", len(drafted), len(to_post), max_per_channel) - # 6. Post to Slack (or dry-run) + # 6. Post to Slack, or (Slack off) write straight to the DB, or dry-run. posted_list: list[dict] = [] slack_client = None + slack_on = False if not dry_run: - from slack_sdk import WebClient - bot_token = getattr(settings, "slack_bot_token_grantbot", "") - if not bot_token or bot_token.startswith("xoxb-placeholder"): - bot_token = settings.slack_bot_token_su - logger.info("No grantbot Slack token — using SuBot's token as fallback") - if bot_token and not bot_token.startswith("xoxb-placeholder"): - slack_client = WebClient(token=bot_token) - _ensure_channel_membership(slack_client, {item.get("channel", channel) for item in to_post}) + from src.services.slack_tokens import slack_globally_enabled + slack_on = await slack_globally_enabled(session) + if slack_on: + from slack_sdk import WebClient + bot_token = getattr(settings, "slack_bot_token_grantbot", "") + if not bot_token or bot_token.startswith("xoxb-placeholder"): + bot_token = settings.slack_bot_token_su + logger.info("No grantbot Slack token — using SuBot's token as fallback") + if bot_token and not bot_token.startswith("xoxb-placeholder"): + slack_client = WebClient(token=bot_token) + _ensure_channel_membership(slack_client, {item.get("channel", channel) for item in to_post}) + else: + logger.info("Slack disabled — GrantBot posting funding opportunities to the DB") for item in to_post: opp = item["opportunity"] @@ -536,9 +566,22 @@ async def _run_grantbot_with_session( logger.info("Skipping FOA %s — already claimed by another run", opp_num) continue + if not slack_on: + # Slack off — write the funding post straight to agent_messages so + # the sim scans it (funding threads are open to all). Keep the claim. + try: + await _post_funding_to_db(session, target_channel, full_post) + logger.info("Posted opportunity %s to #%s (DB)", opp_num, target_channel) + except Exception as exc: + logger.error("Failed to persist %s to #%s: %s", opp_num, target_channel, exc) + await _release_foa(session, opp_num) + continue + posted_list.append({"number": opp_num, "title": title, "channel": target_channel}) + continue + if not slack_client: - # No Slack client available (no token configured). Release the claim - # so a future run with credentials can post this FOA. + # Slack on but no usable token/client. Release the claim so a future + # run with credentials can post this FOA. await _release_foa(session, opp_num) continue diff --git a/src/agent/message_log.py b/src/agent/message_log.py index 2d700c1..50b3c7a 100644 --- a/src/agent/message_log.py +++ b/src/agent/message_log.py @@ -25,6 +25,11 @@ class LogEntry: # memory segment. Default 'public' is safe for all existing callers. # See specs/privacy-and-channel-visibility.md §G2. visibility: str = "public" + # Slack-mirror mapping — set when this message was posted to (or came from) + # Slack. In pure Slack-on mode slack_ts == ts. Persisted to the DB row so the + # reconcile pass can dedup a mirrored message. See specs/local-db-conversations.md. + slack_ts: str | None = None + slack_channel_id: str | None = None def is_funding_post(content: str) -> bool: diff --git a/src/agent/simulation.py b/src/agent/simulation.py index 0c9a8bc..2c1aed8 100644 --- a/src/agent/simulation.py +++ b/src/agent/simulation.py @@ -230,6 +230,10 @@ def __init__( # interface, private-channel handover) enter the simulation. See # _poll_inbound_from_db. self._pi_inbox_cursor: float = 0.0 + # Slack ts values already represented in the DB (canonical id may differ + # if a DB-origin message was later mirrored to Slack). Lets the Slack + # reconcile skip a message it already has. See _rebuild_state_from_slack. + self._known_slack_ts: set[str] = set() # ------------------------------------------------------------------ # Lifecycle @@ -2428,7 +2432,8 @@ async def _post_message( except (TypeError, ValueError): posted_at = time.time() - # Add to message log + # Add to message log. When Slack posted this, record the mirror mapping + # (in pure Slack-on mode slack_ts == ts). entry = LogEntry( ts=ts, channel=channel, @@ -2438,6 +2443,8 @@ async def _post_message( thread_ts=thread_ts, posted_at=posted_at, is_bot=True, + slack_ts=slack_ts, + slack_channel_id=(result.get("channel") if result else None), ) # Persisted to agent_messages via the MessageLog append callback # (_enqueue_persist → _flush_persisted). The DB is the primary store. @@ -2654,12 +2661,14 @@ async def _rebuild_state_from_db(self) -> None: loaded += 1 if entry.posted_at > max_posted: max_posted = entry.posted_at - # Advance the Slack poll cursor for rows that were mirrored, so the - # optional reconcile only fetches genuinely newer Slack messages. - if r.slack_ts and r.slack_channel_id: - cur = self._poll_cursors.get(r.slack_channel_id, "0") - if r.slack_ts > cur: - self._poll_cursors[r.slack_channel_id] = r.slack_ts + # Track the Slack mapping so the reconcile can dedup, and advance + # the Slack poll cursor so it only fetches genuinely newer messages. + if r.slack_ts: + self._known_slack_ts.add(r.slack_ts) + if r.slack_channel_id: + cur = self._poll_cursors.get(r.slack_channel_id, "0") + if r.slack_ts > cur: + self._poll_cursors[r.slack_channel_id] = r.slack_ts self._last_mint_ts = max(self._last_mint_ts, max_posted) # Start the inbox poller past all restored history so it only picks up # genuinely new web-written PI messages. @@ -2701,6 +2710,9 @@ async def _flush_persisted(self) -> None: "sender_name": e.sender_name or "", "is_bot": e.is_bot, "posted_at": e.posted_at, + "slack_ts": e.slack_ts, + "slack_channel_id": e.slack_channel_id, + "slack_thread_ts": e.thread_ts if e.slack_ts else None, } rows = list(by_ts.values()) if not rows: @@ -2724,6 +2736,9 @@ async def _flush_persisted(self) -> None: "channel_id": stmt.excluded.channel_id, "channel_name": stmt.excluded.channel_name, "agent_id": stmt.excluded.agent_id, + "slack_ts": stmt.excluded.slack_ts, + "slack_channel_id": stmt.excluded.slack_channel_id, + "slack_thread_ts": stmt.excluded.slack_thread_ts, }, ) await db.execute(stmt) @@ -2797,6 +2812,13 @@ async def _rebuild_state_from_slack(self) -> None: if is_bot and user_id: sender_agent_id = bot_uid_to_agent.get(user_id) + # Skip messages already represented in the DB (dedup a message + # that was DB-origin then mirrored to Slack, whose canonical id + # differs from this Slack ts). + if ts and ts in self._known_slack_ts: + if ts: + self._poll_cursors[ch_id] = ts + continue sender_name = msg.get("username", "") or user_id entry = LogEntry( ts=ts, @@ -2808,9 +2830,13 @@ async def _rebuild_state_from_slack(self) -> None: posted_at=float(ts) if ts else 0.0, is_bot=is_bot, visibility=ch_visibility, + slack_ts=ts or None, + slack_channel_id=ch_id, ) - self.message_log.append(entry) - total_messages += 1 + if self.message_log.append(entry): + total_messages += 1 + if ts: + self._known_slack_ts.add(ts) # Update poll cursor to latest if ts: @@ -2828,6 +2854,8 @@ async def _rebuild_state_from_slack(self) -> None: rts = reply.get("ts", "") if rts == ts: continue # skip parent (already added) + if rts and rts in self._known_slack_ts: + continue r_user_id = reply.get("user", "") r_is_bot = bool(reply.get("bot_id")) or reply.get("subtype") == "bot_message" r_agent_id = bot_uid_to_agent.get(r_user_id) if r_is_bot else None @@ -2841,9 +2869,13 @@ async def _rebuild_state_from_slack(self) -> None: posted_at=float(rts) if rts else 0.0, is_bot=r_is_bot, visibility=ch_visibility, + slack_ts=rts or None, + slack_channel_id=ch_id, ) - self.message_log.append(r_entry) - total_messages += 1 + if self.message_log.append(r_entry): + total_messages += 1 + if rts: + self._known_slack_ts.add(rts) logger.info( "Slack reconcile: appended %d messages across %d channels, %d threads", diff --git a/src/routers/agent_page.py b/src/routers/agent_page.py index 60a3fe0..d22dc30 100644 --- a/src/routers/agent_page.py +++ b/src/routers/agent_page.py @@ -583,41 +583,53 @@ async def reopen_proposal( # Legacy fallback: flag is off → post guidance verbatim to the origin # public thread. This reproduces the pre-refactor behavior and is the # same code as before; kept gated so rollback is a config change. - try: - from slack_sdk import WebClient - - from src.services.slack_tokens import token_for_agent_row - bot_token = token_for_agent_row(agent) - if not bot_token: - raise HTTPException(status_code=500, detail="No bot token available") - client = WebClient(token=bot_token) - channels_result = client.conversations_list( - types="public_channel,private_channel", limit=200, - ) - channel_id = None - for ch in channels_result.get("channels", []): - if ch["name"] == td.channel: - channel_id = ch["id"] - break - if not channel_id: - raise HTTPException(status_code=500, detail=f"Channel #{td.channel} not found") - client.chat_postMessage( - channel=channel_id, - text=f"*PI guidance from {current_user.name}:*\n\n{guidance}", - thread_ts=td.thread_id, - ) - logger.warning( - "LEGACY PATH: PI %s posted guidance in proposal thread %s via %s " - "(enable_private_refinement=False)", - current_user.name, td.thread_id, agent.agent_id, - ) - except HTTPException: - raise - except Exception as exc: - logger.error("Failed to post PI guidance to Slack: %s", exc) - raise HTTPException( - status_code=500, detail=f"Failed to post to Slack: {str(exc)[:100]}", - ) + from src.services.slack_tokens import slack_globally_enabled, token_for_agent_row + + if not await slack_globally_enabled(db): + # Slack off → write the guidance to the DB inbox on the origin thread. + from src.services.pi_inbox import get_latest_run_id, record_pi_message + run_id = await get_latest_run_id(db) + if run_id: + await record_pi_message( + db, run_id=run_id, channel_name=td.channel, + content=f"PI guidance from {current_user.name}: {guidance}", + sender_name=f"{current_user.name} (PI)", thread_ts=td.thread_id, + ) + logger.info("Reopen guidance for %s written to DB inbox (Slack off)", td.thread_id) + else: + try: + from slack_sdk import WebClient + bot_token = token_for_agent_row(agent) + if not bot_token: + raise HTTPException(status_code=500, detail="No bot token available") + client = WebClient(token=bot_token) + channels_result = client.conversations_list( + types="public_channel,private_channel", limit=200, + ) + channel_id = None + for ch in channels_result.get("channels", []): + if ch["name"] == td.channel: + channel_id = ch["id"] + break + if not channel_id: + raise HTTPException(status_code=500, detail=f"Channel #{td.channel} not found") + client.chat_postMessage( + channel=channel_id, + text=f"*PI guidance from {current_user.name}:*\n\n{guidance}", + thread_ts=td.thread_id, + ) + logger.warning( + "LEGACY PATH: PI %s posted guidance in proposal thread %s via %s " + "(enable_private_refinement=False)", + current_user.name, td.thread_id, agent.agent_id, + ) + except HTTPException: + raise + except Exception as exc: + logger.error("Failed to post PI guidance to Slack: %s", exc) + raise HTTPException( + status_code=500, detail=f"Failed to post to Slack: {str(exc)[:100]}", + ) existing = await db.execute( select(ProposalReview).where( diff --git a/src/services/email_inbound.py b/src/services/email_inbound.py index 9e86ef2..7dbe7f3 100644 --- a/src/services/email_inbound.py +++ b/src/services/email_inbound.py @@ -504,9 +504,25 @@ async def _handle_instruction( else: # Legacy fallback: flag off → post guidance verbatim to the origin # public thread (same behavior as the web legacy path). - from slack_sdk import WebClient + from src.services.slack_tokens import slack_globally_enabled, token_for_agent_row + + # Slack off → write the guidance to the DB inbox on the origin thread + # instead of posting to Slack. + if not await slack_globally_enabled(db): + from src.services.pi_inbox import get_latest_run_id, record_pi_message + run_id = await get_latest_run_id(db) + if run_id: + await record_pi_message( + db, run_id=run_id, channel_name=td.channel, + content=f"PI guidance from {user.name} (via email): {instruction}", + sender_name=f"{user.name} (PI)", thread_ts=td.thread_id, + ) + logger.info("Email guidance for %s written to DB inbox (Slack off)", td.thread_id) + return True + logger.error("No simulation run to record email guidance for %s", td.thread_id) + return False - from src.services.slack_tokens import token_for_agent_row + from slack_sdk import WebClient bot_token = token_for_agent_row(agent) if not bot_token: logger.error("No bot token for agent %s", agent.agent_id) diff --git a/src/services/slack_tokens.py b/src/services/slack_tokens.py index d226db7..2841931 100644 --- a/src/services/slack_tokens.py +++ b/src/services/slack_tokens.py @@ -49,6 +49,20 @@ async def get_agent_bot_token(db: AsyncSession, agent_id: str) -> str | None: return env_token(agent_id) +async def slack_globally_enabled(db: AsyncSession) -> bool: + """Whether Slack integration is on for this deployment. + + Explicit SLACK_ENABLED wins; otherwise auto-detect (on iff at least one + usable bot token exists anywhere). Used to gate secondary Slack posters + (GrantBot, the email→Slack relay, web-triggered posts) so they no-op in + DB-only mode. See specs/local-db-conversations.md. + """ + setting = get_settings().slack_enabled + if setting is not None: + return setting + return await get_any_bot_token(db) is not None + + async def get_any_bot_token(db: AsyncSession) -> str | None: """Any valid bot token, for workspace-wide lookups (e.g. users.lookupByEmail). From 33d31cbea8f85865d48234b30db32a340199fc9b Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Mon, 20 Jul 2026 16:23:15 -0700 Subject: [PATCH 06/10] Stage 7: DM persistence (migration 0020) + web DMs + one-time Slack backfill MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the DB-primary refactor: PI<->bot DMs are durable and Slack-optional, and there's a one-time importer for existing Slack history. DM persistence: - Migration 0020 + PiDmMessage model: pi_dm_messages (run, agent_id, pi_user_id, direction inbound/outbound, content, ts, slack_ts, posted_at). pi_user_id is a Slack user id (Slack-on) or local: (web). - pi_inbox.record_pi_dm() / web_pi_user_id(); PIHandler._send_dm now records every outbound DM (durable + visible in the web UI even with Slack off) and takes simulation_run_id. - DM processing is unified through the DB: _poll_pi_dms (Slack) now only RECORDS inbound DMs as rows; the new _poll_pi_dms_from_db is the single processor that runs each inbound row through PIHandler.handle_dm and flips has_pi_directive. This handles Slack and web DMs identically with no double-processing. _seed_pi_dm_cursor starts it past existing history on boot. PI web interface (DMs): POST /agent/{id}/dm writes an inbound DM row; the conversations page shows the recent DM thread and a send box. Ops: - --fresh now also wipes pi_dm_messages. - scripts/backfill_slack_history_to_db.py: one-time importer that reuses the engine's setup + Slack reconcile + flush to pull channel/thread history (content + slack_ts) into agent_messages for a run, without running any turns. Verification: full suite 314 passed; a real-Postgres smoke test (inbound web DM recorded → processed exactly once → directive flag set → re-poll no-op; outbound DM persisted); app/worker/grantbot restarted cleanly; backfill script imports. Co-Authored-By: Claude Opus 4.8 (1M context) --- alembic/versions/0020_pi_dm_messages.py | 66 +++++++++++++ scripts/backfill_slack_history_to_db.py | 125 ++++++++++++++++++++++++ src/agent/main.py | 3 +- src/agent/pi_handler.py | 29 +++++- src/agent/simulation.py | 100 ++++++++++++++++--- src/models/__init__.py | 2 + src/models/agent_activity.py | 44 +++++++++ src/routers/agent_page.py | 52 +++++++++- src/services/pi_inbox.py | 35 ++++++- templates/agent/conversations.html | 21 ++++ 10 files changed, 455 insertions(+), 22 deletions(-) create mode 100644 alembic/versions/0020_pi_dm_messages.py create mode 100644 scripts/backfill_slack_history_to_db.py diff --git a/alembic/versions/0020_pi_dm_messages.py b/alembic/versions/0020_pi_dm_messages.py new file mode 100644 index 0000000..b2acd57 --- /dev/null +++ b/alembic/versions/0020_pi_dm_messages.py @@ -0,0 +1,66 @@ +"""Add pi_dm_messages table (durable PI<->bot direct messages) + +Revision ID: 0020 +Revises: 0019 +Create Date: 2026-07-20 00:00:00.000000 + +DMs never entered the shared message log, so they had no durable home. This +table stores them so a PI can DM their bot (standing instructions, questions) +with Slack fully off. See specs/local-db-conversations.md. +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +revision: str = "0020" +down_revision: Union[str, None] = "0019" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "pi_dm_messages", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column( + "simulation_run_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("simulation_runs.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("agent_id", sa.String(50), nullable=False), + sa.Column("pi_user_id", sa.String(50), nullable=False), + sa.Column( + "direction", + sa.Enum("inbound", "outbound", name="pi_dm_direction_enum"), + nullable=False, + ), + sa.Column("content", sa.Text(), nullable=False), + sa.Column("sender_name", sa.String(100), nullable=False, server_default=""), + sa.Column("ts", sa.String(50), nullable=False), + sa.Column("slack_ts", sa.String(50), nullable=True), + sa.Column("posted_at", sa.Float(), nullable=False, server_default="0"), + sa.Column( + "created_at", sa.DateTime(timezone=True), + server_default=sa.func.now(), nullable=False, + ), + ) + op.create_index( + "ix_pi_dm_run_agent_posted", "pi_dm_messages", + ["simulation_run_id", "agent_id", "posted_at"], + ) + op.create_index( + "ix_pi_dm_run_direction_posted", "pi_dm_messages", + ["simulation_run_id", "direction", "posted_at"], + ) + + +def downgrade() -> None: + op.drop_index("ix_pi_dm_run_direction_posted", table_name="pi_dm_messages") + op.drop_index("ix_pi_dm_run_agent_posted", table_name="pi_dm_messages") + op.drop_table("pi_dm_messages") + sa.Enum(name="pi_dm_direction_enum").drop(op.get_bind(), checkfirst=True) diff --git a/scripts/backfill_slack_history_to_db.py b/scripts/backfill_slack_history_to_db.py new file mode 100644 index 0000000..d0762d9 --- /dev/null +++ b/scripts/backfill_slack_history_to_db.py @@ -0,0 +1,125 @@ +"""One-time backfill: import current Slack conversation history into the DB. + +Since the DB became the primary conversation store (specs/local-db-conversations.md), +agent_messages carries message content. Historically content lived only in Slack, +so pre-cutover runs have metadata-only rows. Run this once, with Slack tokens +available, to pull the workspace's channel + thread history into agent_messages +(content + slack_ts as the canonical message_ts) before switching to DB-primary +operation, preserving in-flight conversations. + +It reuses the engine's own setup/rebuild machinery (seeded + private channels, +the Slack reconcile, and the persist flush), then exits — it does NOT run any +agent turns or make LLM calls. + +Idempotent: the reconcile appends only messages not already in the DB, and the +flush upserts on (simulation_run_id, message_ts). Safe to re-run. + +Usage (inside the app container): + + docker exec copi-python-opus-app-1 python scripts/backfill_slack_history_to_db.py +""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from sqlalchemy import desc, func, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from src.agent.agent import Agent +from src.agent.simulation import SimulationEngine +from src.config import get_settings +from src.models import AgentMessage, AgentRegistry, SimulationRun +from src.services.slack_tokens import env_token, is_valid_token + +logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") +logger = logging.getLogger("backfill_slack_history") + + +async def main(run_id_arg: str | None) -> None: + settings = get_settings() + engine = create_async_engine(settings.database_url) + sf = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + # Roster (active agents) + tokens, mirroring src/agent/main.py. + async with sf() as db: + rows = (await db.execute( + select( + AgentRegistry.agent_id, AgentRegistry.bot_name, + AgentRegistry.pi_name, AgentRegistry.slack_bot_token, + ).where(AgentRegistry.status == "active").order_by(AgentRegistry.agent_id) + )).all() + if run_id_arg: + run_id = run_id_arg + else: + run_id = (await db.execute( + select(SimulationRun.id).order_by(desc(SimulationRun.started_at)).limit(1) + )).scalar_one_or_none() + + if run_id is None: + logger.error("No SimulationRun found — start a run first (nothing to attach to).") + await engine.dispose() + return + + agents = [Agent(agent_id=r.agent_id, bot_name=r.bot_name, pi_name=r.pi_name) for r in rows] + + from src.agent.slack_client import AgentSlackClient + slack_clients = {} + for r in rows: + tok = r.slack_bot_token if is_valid_token(r.slack_bot_token) else env_token(r.agent_id) + if is_valid_token(tok): + client = AgentSlackClient(agent_id=r.agent_id, bot_token=tok) + if client.connect(): + slack_clients[r.agent_id] = client + if not slack_clients: + logger.error("No connected Slack clients — cannot backfill from Slack.") + await engine.dispose() + return + + async with sf() as db: + before = (await db.execute( + select(func.count(AgentMessage.id)).where( + AgentMessage.simulation_run_id == run_id, + func.length(AgentMessage.content) > 0, + ) + )).scalar_one() + + sim = SimulationEngine( + agents=agents, slack_clients=slack_clients, session_factory=sf, + simulation_run_id=run_id, slack_enabled=True, + ) + # Reuse the engine's setup + rebuild, then flush to the DB. No turns run. + sim._ensure_seeded_channels() + await sim._persist_seeded_channels() + await sim._sync_private_channels_from_db() + sim.message_log.set_persist_callback(sim._enqueue_persist) + await sim._rebuild_state_from_db() + await sim._rebuild_state_from_slack() + await sim._flush_persisted() + + async with sf() as db: + after = (await db.execute( + select(func.count(AgentMessage.id)).where( + AgentMessage.simulation_run_id == run_id, + func.length(AgentMessage.content) > 0, + ) + )).scalar_one() + + logger.info( + "Backfill complete for run %s: content-bearing messages %d -> %d (log holds %d).", + run_id, before, after, len(sim.message_log), + ) + await engine.dispose() + + +if __name__ == "__main__": + ap = argparse.ArgumentParser() + ap.add_argument("--run-id", default=None, help="Target SimulationRun id (default: latest)") + args = ap.parse_args() + asyncio.run(main(args.run_id)) diff --git a/src/agent/main.py b/src/agent/main.py index 98d219a..60813a1 100644 --- a/src/agent/main.py +++ b/src/agent/main.py @@ -140,7 +140,7 @@ def _token_for(agent_id: str) -> str | None: if not no_db: from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine - from src.models import AgentChannel, AgentMessage, SimulationRun + from src.models import AgentChannel, AgentMessage, PiDmMessage, SimulationRun engine = create_async_engine(settings.database_url) session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) @@ -151,6 +151,7 @@ def _token_for(agent_id: str) -> str | None: async with session_factory() as db: await db.execute(AgentMessage.__table__.delete()) await db.execute(AgentChannel.__table__.delete()) + await db.execute(PiDmMessage.__table__.delete()) await db.commit() logger.info("Simulation data wiped.") diff --git a/src/agent/pi_handler.py b/src/agent/pi_handler.py index f4ec7cf..09eb104 100644 --- a/src/agent/pi_handler.py +++ b/src/agent/pi_handler.py @@ -28,6 +28,7 @@ def __init__( pi_slack_id_to_agent_ids: dict[str, list[str]], message_log: MessageLog, session_factory=None, + simulation_run_id=None, ): self.agents = agents self.slack_clients = slack_clients @@ -40,6 +41,7 @@ def __init__( } self.message_log = message_log self.session_factory = session_factory + self.simulation_run_id = simulation_run_id # ------------------------------------------------------------------ # DM handling @@ -369,12 +371,33 @@ async def notify_thread_conclusion( # ------------------------------------------------------------------ async def _send_dm(self, agent_id: str, pi_slack_id: str, text: str) -> None: - """Send a DM from the agent's bot to the PI.""" + """Send a DM from the agent's bot to the PI (Slack + DB record). + + Persists an outbound row so the DM is durable and visible in the web UI + even when Slack is off. See specs/local-db-conversations.md. + """ client = self.slack_clients.get(agent_id) + slack_ts = None if client and client.is_connected: - client.send_dm(pi_slack_id, text) + result = client.send_dm(pi_slack_id, text) + if isinstance(result, dict): + slack_ts = result.get("ts") else: - logger.debug("[%s] Cannot send DM — no connected client", agent_id) + logger.debug("[%s] Cannot send DM via Slack — recording to DB only", agent_id) + + if self.session_factory and self.simulation_run_id: + try: + from src.services.pi_inbox import record_pi_dm + agent = self.agents.get(agent_id) + async with self.session_factory() as db: + await record_pi_dm( + db, run_id=self.simulation_run_id, agent_id=agent_id, + pi_user_id=pi_slack_id, direction="outbound", content=text, + sender_name=agent.bot_name if agent else agent_id, slack_ts=slack_ts, + ) + await db.commit() + except Exception as exc: + logger.debug("[%s] Could not record outbound DM: %s", agent_id, exc) @staticmethod def _parse_json(text: str) -> dict: diff --git a/src/agent/simulation.py b/src/agent/simulation.py index 2c1aed8..2babe59 100644 --- a/src/agent/simulation.py +++ b/src/agent/simulation.py @@ -234,6 +234,9 @@ def __init__( # if a DB-origin message was later mirrored to Slack). Lets the Slack # reconcile skip a message it already has. See _rebuild_state_from_slack. self._known_slack_ts: set[str] = set() + # High-water mark (posted_at) for the DB DM inbox poller (Slack-off / + # web PI DMs). See _poll_pi_dms_from_db. + self._pi_dm_cursor: float = 0.0 # ------------------------------------------------------------------ # Lifecycle @@ -309,6 +312,7 @@ async def start(self) -> None: await self._rebuild_state_from_db() await self._rebuild_state_from_slack() await self._rebuild_agent_state() + await self._seed_pi_dm_cursor() # Rebuild advanced last_seen_cursor to max(all_messages), which can # overshoot messages in private channels (typically older than the # latest public chatter). Rewind member-bot cursors so Phase 2 can @@ -327,6 +331,7 @@ async def start(self) -> None: pi_slack_id_to_agent_ids=self._pi_slack_id_to_agent_ids, message_log=self.message_log, session_factory=self.session_factory, + simulation_run_id=self.simulation_run_id, ) # Main loop @@ -343,6 +348,9 @@ async def start(self) -> None: # web interface, private-channel handover). Runs regardless of Slack, # and is how PIs interact when Slack is off. await self._poll_inbound_from_db() + # DB-native PI DM processing (Slack DMs recorded by _poll_pi_dms and + # web DMs both converge here). + await self._poll_pi_dms_from_db() # Sync proposal reviews and any newly-created private channels from # the web app. Both are DB-driven, so a single tick picks them up. @@ -2216,13 +2224,21 @@ async def _reopen_thread(self, agent_id: str, thread_ts: str, pi_entry: LogEntry logger.info("[%s] PI reopened closed thread %s with %s", agent_id, thread_ts, other_id) async def _poll_pi_dms(self) -> None: - """Poll for DMs from PIs and process them via PIHandler.""" - if not self._pi_handler or not self._pi_slack_id_to_agent_ids: + """Poll Slack for PI DMs and record them as inbound rows. + + Processing is unified through the DB: this method only persists inbound + Slack DMs to pi_dm_messages; _poll_pi_dms_from_db is the single place + that runs them through PIHandler (so Slack and web DMs are handled + identically and never double-processed). See specs/local-db-conversations.md. + """ + if not self._pi_slack_id_to_agent_ids or not self.session_factory or not self.simulation_run_id: return # Default cursor to simulation start time — only process DMs sent after we started default_cursor = str(self._start_time.timestamp()) if self._start_time else "0" + from src.services.pi_inbox import record_pi_dm + for pi_slack_id, agent_ids in self._pi_slack_id_to_agent_ids.items(): for agent_id in agent_ids: client = self.slack_clients.get(agent_id) @@ -2237,26 +2253,78 @@ async def _poll_pi_dms(self) -> None: text = msg.get("text", "").strip() if not text: continue - logger.info("[%s] PI DM from %s: %s", agent_id, pi_slack_id, text[:80]) - try: - await self._pi_handler.handle_dm(agent_id, pi_slack_id, text) - # PI DMs deliberately do not update public working memory: - # standing instructions are persisted to the private profile - # by _handle_standing_instruction; other DM categories are - # handled in-band. has_pi_directive still flips so Phase 5 - # runs this turn. - agent = self.agents.get(agent_id) - if agent: - agent.state.has_pi_directive = True + async with self.session_factory() as db: + await record_pi_dm( + db, run_id=self.simulation_run_id, agent_id=agent_id, + pi_user_id=pi_slack_id, direction="inbound", content=text, + sender_name="PI", slack_ts=ts or None, + ) + await db.commit() except Exception as exc: - logger.error("[%s] Failed to handle PI DM: %s", agent_id, exc, exc_info=True) - - # Update cursor to this message + logger.error("[%s] Failed to record PI DM: %s", agent_id, exc) if ts > oldest: self._dm_poll_cursors[agent_id] = ts + async def _seed_pi_dm_cursor(self) -> None: + """Start the DM poller past existing inbound DMs (don't replay history).""" + if not self.session_factory or not self.simulation_run_id: + return + from sqlalchemy import func as sa_func + from sqlalchemy import select as sa_select + from src.models import PiDmMessage + try: + async with self.session_factory() as db: + mx = (await db.execute( + sa_select(sa_func.max(PiDmMessage.posted_at)).where( + PiDmMessage.simulation_run_id == self.simulation_run_id, + PiDmMessage.direction == "inbound", + ) + )).scalar_one_or_none() + if mx: + self._pi_dm_cursor = max(self._pi_dm_cursor, mx) + except Exception as exc: + logger.debug("PI DM cursor seed failed: %s", exc) + + async def _poll_pi_dms_from_db(self) -> None: + """Process inbound PI DMs recorded in the DB (Slack or web-originated). + + The single processor for PI DMs: reads new inbound pi_dm_messages rows + and runs each through PIHandler.handle_dm (classify → standing + instruction / feedback / question), then flips has_pi_directive so + Phase 5 runs. Works with Slack off. See specs/local-db-conversations.md. + """ + if not self._pi_handler or not self.session_factory or not self.simulation_run_id: + return + from sqlalchemy import select as sa_select + from src.models import PiDmMessage + try: + async with self.session_factory() as db: + rows = (await db.execute( + sa_select(PiDmMessage) + .where( + PiDmMessage.simulation_run_id == self.simulation_run_id, + PiDmMessage.direction == "inbound", + PiDmMessage.posted_at > self._pi_dm_cursor, + ) + .order_by(PiDmMessage.posted_at.asc()) + )).scalars().all() + except Exception as exc: + logger.debug("PI DM inbox poll failed: %s", exc) + return + + for r in rows: + if r.posted_at > self._pi_dm_cursor: + self._pi_dm_cursor = r.posted_at + if r.agent_id not in self.agents: + continue + try: + await self._pi_handler.handle_dm(r.agent_id, r.pi_user_id, r.content) + self.agents[r.agent_id].state.has_pi_directive = True + except Exception as exc: + logger.error("[%s] Failed to handle PI DM (DB): %s", r.agent_id, exc) + async def _poll_proposal_threads_for_pi(self) -> None: """Poll unreviewed proposal threads for PI replies. diff --git a/src/models/__init__.py b/src/models/__init__.py index f9e12d2..d9b95d1 100644 --- a/src/models/__init__.py +++ b/src/models/__init__.py @@ -8,6 +8,7 @@ AgentChannel, AgentMessage, LlmCallLog, + PiDmMessage, PrivateChannelMember, SimulationRun, ThreadDecision, @@ -40,6 +41,7 @@ "AgentChannel", "LlmCallLog", "ThreadDecision", + "PiDmMessage", "PrivateChannelMember", "VISIBILITY_PUBLIC", "VISIBILITY_COLLAB_PRIVATE", diff --git a/src/models/agent_activity.py b/src/models/agent_activity.py index 8f773d4..7c9beab 100644 --- a/src/models/agent_activity.py +++ b/src/models/agent_activity.py @@ -293,3 +293,47 @@ class PrivateChannelMember(Base): def __repr__(self) -> str: who = f"agent={self.agent_id}" if self.agent_id else f"user={self.user_id}" return f"" + + +class PiDmMessage(Base): + """A direct message between a PI (human) and their agent's bot. + + DMs never enter the shared MessageLog, so they get their own durable home + here (the DB is the primary store, not Slack). Inbound rows (direction= + 'inbound') are written by the Slack DM poller or the PI web interface and + ingested by SimulationEngine._poll_pi_dms_from_db; outbound rows record + what the bot sent back. See specs/local-db-conversations.md. + """ + + __tablename__ = "pi_dm_messages" + + id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 + ) + simulation_run_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("simulation_runs.id", ondelete="CASCADE"), + nullable=False, + ) + agent_id: Mapped[str] = mapped_column(String(50), nullable=False) + # PI identity: Slack user id (Slack-on) or "local:" (Slack-off). + pi_user_id: Mapped[str] = mapped_column(String(50), nullable=False) + direction: Mapped[str] = mapped_column( + Enum("inbound", "outbound", name="pi_dm_direction_enum"), nullable=False + ) + content: Mapped[str] = mapped_column(Text, nullable=False) + sender_name: Mapped[str] = mapped_column(String(100), nullable=False, server_default="") + ts: Mapped[str] = mapped_column(String(50), nullable=False) # canonical id + slack_ts: Mapped[str | None] = mapped_column(String(50), nullable=True) + posted_at: Mapped[float] = mapped_column(Float, nullable=False, server_default="0") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + + __table_args__ = ( + Index("ix_pi_dm_run_agent_posted", "simulation_run_id", "agent_id", "posted_at"), + Index("ix_pi_dm_run_direction_posted", "simulation_run_id", "direction", "posted_at"), + ) + + def __repr__(self) -> str: + return f"" diff --git a/src/routers/agent_page.py b/src/routers/agent_page.py index d22dc30..caa9b66 100644 --- a/src/routers/agent_page.py +++ b/src/routers/agent_page.py @@ -688,7 +688,22 @@ async def agent_conversations( run_id = await get_latest_run_id(db) channels: list[str] = [] messages: list[dict] = [] + dms: list[dict] = [] if run_id: + from src.models import PiDmMessage + dm_rows = await db.execute( + select(PiDmMessage) + .where( + PiDmMessage.simulation_run_id == run_id, + PiDmMessage.agent_id == aid, + ) + .order_by(PiDmMessage.posted_at.desc()) + .limit(20) + ) + dms = [ + {"direction": d.direction, "sender": d.sender_name or "", "content": d.content} + for d in reversed(dm_rows.scalars().all()) + ] # Channels this agent participates in (has authored a message in). ch_rows = await db.execute( select(distinct(AgentMessage.channel_name)).where( @@ -726,7 +741,8 @@ async def agent_conversations( "agent/conversations.html", _template_context( request, current_user, agent=agent, is_owner=is_owner, - channels=channels, messages=messages, has_run=run_id is not None, + channels=channels, messages=messages, dms=dms, + has_run=run_id is not None, posted=request.query_params.get("posted"), ), ) @@ -779,6 +795,40 @@ async def post_agent_message( return RedirectResponse(url=f"/agent/{agent_id}/conversations?posted=1", status_code=302) +@router.post("/{agent_id}/dm") +async def send_agent_dm( + agent_id: str, + request: Request, + content: str = Form(...), + db: AsyncSession = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Send a DM directive to the agent's bot (standing instruction / question). + + Writes an inbound pi_dm_messages row; the sim processes it via + _poll_pi_dms_from_db (same path as a Slack DM). See specs/local-db-conversations.md. + """ + from src.services.pi_inbox import get_latest_run_id, record_pi_dm, web_pi_user_id + + agent, is_owner = await get_agent_with_access(agent_id, db, current_user) + if agent.status != "active": + raise HTTPException(status_code=403, detail="Agent is not active") + text = content.strip() + if not text: + raise HTTPException(status_code=400, detail="Message cannot be empty") + run_id = await get_latest_run_id(db) + if not run_id: + raise HTTPException(status_code=409, detail="No simulation run yet") + await record_pi_dm( + db, run_id=run_id, agent_id=agent_id, + pi_user_id=web_pi_user_id(current_user.id), direction="inbound", + content=text, sender_name=f"{current_user.name} (PI)", + ) + await db.commit() + logger.info("[%s] PI %s sent a web DM directive", agent_id, current_user.name) + return RedirectResponse(url=f"/agent/{agent_id}/conversations?posted=1", status_code=302) + + @router.get("/{agent_id}/profile", response_class=HTMLResponse) async def view_private_profile( agent_id: str, diff --git a/src/services/pi_inbox.py b/src/services/pi_inbox.py index 8f48fb3..c32ab2b 100644 --- a/src/services/pi_inbox.py +++ b/src/services/pi_inbox.py @@ -13,7 +13,7 @@ from sqlalchemy import desc, select from sqlalchemy.ext.asyncio import AsyncSession -from src.models import AgentChannel, AgentMessage, SimulationRun +from src.models import AgentChannel, AgentMessage, PiDmMessage, SimulationRun async def get_latest_run_id(db: AsyncSession) -> uuid.UUID | None: @@ -76,3 +76,36 @@ async def record_pi_message( ) db.add(msg) return msg + + +async def record_pi_dm( + db: AsyncSession, + *, + run_id: uuid.UUID, + agent_id: str, + pi_user_id: str, + direction: str, # 'inbound' (PI→bot) or 'outbound' (bot→PI) + content: str, + sender_name: str = "", + slack_ts: str | None = None, +) -> PiDmMessage: + """Persist a PI<->bot direct message. Does not commit.""" + ts = f"{time.time():.6f}" + dm = PiDmMessage( + simulation_run_id=run_id, + agent_id=agent_id, + pi_user_id=pi_user_id, + direction=direction, + content=content, + sender_name=sender_name, + ts=ts, + slack_ts=slack_ts, + posted_at=float(ts), + ) + db.add(dm) + return dm + + +def web_pi_user_id(user_id: uuid.UUID) -> str: + """Stable pi_user_id for a web (Slack-off) PI: ``local:``.""" + return f"local:{user_id}" diff --git a/templates/agent/conversations.html b/templates/agent/conversations.html index e1af39d..008d290 100644 --- a/templates/agent/conversations.html +++ b/templates/agent/conversations.html @@ -50,6 +50,27 @@

{{ agent.bot_name }} — Conver + +
+

Direct messages

+

Send a standing instruction ("always…", "never…") or a question. Your bot handles it like a Slack DM.

+ {% if dms %} +
+ {% for d in dms %} +
+ {{ d.content }} +
+ {% endfor %} +
+ {% endif %} +
+ + +
+
+

Recent activity

{% if messages %} From 771b65590b666f0afcf2d1cdd9fe78c4d4543876 Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Tue, 21 Jul 2026 09:21:14 -0700 Subject: [PATCH 07/10] Fix: import AgentChannel in simulation (Stage 2 _persist_seeded_channels) _persist_seeded_channels referenced AgentChannel but it wasn't in the module import, raising NameError ("Failed to persist seeded channels") on startup so seeded channels were never recorded in agent_channels. Surfaced by the first Slack-off agent run. Add AgentChannel to the src.models import. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/agent/simulation.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agent/simulation.py b/src/agent/simulation.py index 2babe59..14e037b 100644 --- a/src/agent/simulation.py +++ b/src/agent/simulation.py @@ -27,7 +27,7 @@ from src.agent.state import PostRef, ProposalRef, ThreadState from src.agent.tools import TOOL_DEFINITIONS, execute_tool from src.config import get_settings -from src.models import AgentMessage, LlmCallLog, ProposalReview, SimulationRun, ThreadDecision +from src.models import AgentChannel, AgentMessage, LlmCallLog, ProposalReview, SimulationRun, ThreadDecision from src.models.agent_activity import VISIBILITY_COLLAB_PRIVATE, VISIBILITY_PUBLIC from src.services.llm import ( generate_agent_response, From 09041a1dd9ba71e631b83c9be9ac7609214b6e36 Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Tue, 21 Jul 2026 10:34:47 -0700 Subject: [PATCH 08/10] Populate slack_ts on inbound-polled human messages The channel poller and proposal-thread PI-reply poller recorded human Slack messages with message_ts set to the Slack ts but left the slack_ts / slack_channel_id mirror-mapping columns null, so "origin = Slack" reporting (and the reconcile's slack_ts dedup set) didn't see them. Content and dedup were unaffected (dedup keys on message_ts, which already equals the Slack ts here), but the mapping is now recorded for consistency with agent posts and the bulk reconcile. Web-origin reopen guidance stays null (it is DB-origin, not Slack). Verification: full suite 314 passed. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/agent/simulation.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/agent/simulation.py b/src/agent/simulation.py index 14e037b..417ab4d 100644 --- a/src/agent/simulation.py +++ b/src/agent/simulation.py @@ -2030,6 +2030,8 @@ async def _poll_slack_for_pi_messages(self) -> None: posted_at=float(ts) if ts else 0.0, is_bot=False, visibility=ch_visibility, + slack_ts=ts or None, + slack_channel_id=ch_id, ) self.message_log.append(entry) logger.info( @@ -2415,6 +2417,8 @@ async def _poll_proposal_threads_for_pi(self) -> None: thread_ts=thread_id, posted_at=float(ts) if ts else 0.0, is_bot=False, + slack_ts=ts or None, + slack_channel_id=ch_id, ) # Avoid re-processing messages already in the log From 311deba4e7c8364cb193f961fde001c93d11e685 Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Tue, 21 Jul 2026 13:04:20 -0700 Subject: [PATCH 09/10] Align db-primary branch to the new test-suite layout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merged main (the test-suite-characterization overhaul: tests/unit|integration| characterization|contract split, testcontainers harness, fakes/factories, CI gate). No conflicts with the refactor's src/ changes. Alignment fixups: - Move tests/test_transport.py -> tests/unit/ to match the new layout (still collected before, but keeps the convention). My edits to the three relocated files (test_message_log, test_simulation_logic, test_private_channel_migration) came across cleanly via git's rename-follow. - test_simulation_logic: add strict=False to the mint_ts zip() (ruff B905; the test lint gate is enforced by scripts/ci.sh). - test_harness_smoke: bump the pinned alembic head 0018 -> 0020 (this branch adds migrations 0019 + 0020). Verified: scripts/ci.sh passes — 460 tests (unit+integration+characterization+ contract), 13 golden-master snapshots unchanged, coverage 35.62% >= 35% floor, ruff clean; migrations 0019/0020 apply via the real alembic chain in the integration harness. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/integration/test_harness_smoke.py | 2 +- tests/unit/test_simulation_logic.py | 2 +- tests/{ => unit}/test_transport.py | 0 3 files changed, 2 insertions(+), 2 deletions(-) rename tests/{ => unit}/test_transport.py (100%) diff --git a/tests/integration/test_harness_smoke.py b/tests/integration/test_harness_smoke.py index 380a113..b60bb37 100644 --- a/tests/integration/test_harness_smoke.py +++ b/tests/integration/test_harness_smoke.py @@ -7,7 +7,7 @@ async def test_container_is_migrated(engine): async with engine.connect() as conn: v = (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalar_one() - assert v == "0018" + assert v == "0020" # bumped by db-primary-conversations migrations 0019 + 0020 async def test_writes_are_rolled_back_part1(db_session): diff --git a/tests/unit/test_simulation_logic.py b/tests/unit/test_simulation_logic.py index a7167d2..ce83801 100644 --- a/tests/unit/test_simulation_logic.py +++ b/tests/unit/test_simulation_logic.py @@ -656,7 +656,7 @@ def test_monotonic_and_unique_under_tight_loop(self): ids = [engine.mint_ts() for _ in range(1000)] floats = [float(x) for x in ids] # Strictly increasing (so posted_at=float(ts) ordering is preserved) - assert all(b > a for a, b in zip(floats, floats[1:])) + assert all(b > a for a, b in zip(floats, floats[1:], strict=False)) # All unique assert len(set(ids)) == len(ids) diff --git a/tests/test_transport.py b/tests/unit/test_transport.py similarity index 100% rename from tests/test_transport.py rename to tests/unit/test_transport.py From 1bfb17010db93f6cdd214e62eb514e2747b465cc Mon Sep 17 00:00:00 2001 From: Mohammad Alanjary Date: Tue, 21 Jul 2026 13:32:16 -0700 Subject: [PATCH 10/10] test: integration coverage for the DB-native PI inbox MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds tests/integration/test_pi_inbox.py (real migrated Postgres) covering src/services/pi_inbox.py — the Slack-independent path by which PI web messages and DMs enter the simulation: - get_latest_run_id picks the most recent run - record_pi_message writes a human row (is_bot=false, agent_id NULL), resolves channel_id/visibility from agent_channels, falls back to local: + public, and sets phase new_post vs thread_reply - record_pi_dm writes inbound/outbound rows; web_pi_user_id formatting Lifts src coverage 35.62% -> 35.86% (pi_inbox was 0%). Full gate: 464 passed, 13 golden-master snapshots unchanged, ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/integration/test_pi_inbox.py | 98 ++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) create mode 100644 tests/integration/test_pi_inbox.py diff --git a/tests/integration/test_pi_inbox.py b/tests/integration/test_pi_inbox.py new file mode 100644 index 0000000..deadedd --- /dev/null +++ b/tests/integration/test_pi_inbox.py @@ -0,0 +1,98 @@ +"""Integration tests for the DB-native PI inbox (src/services/pi_inbox.py). + +These helpers are how a PI's web-authored messages and DMs enter the simulation +when Slack is off — the engine ingests the rows they write. Exercised against the +real migrated Postgres so the actual agent_messages / pi_dm_messages schema +(including the 0019/0020 columns) is validated. See specs/local-db-conversations.md. +""" + +import uuid +from datetime import UTC, datetime, timedelta + +import pytest +from sqlalchemy import select + +from src.models import AgentMessage, PiDmMessage +from src.services.pi_inbox import ( + get_latest_run_id, + record_pi_dm, + record_pi_message, + web_pi_user_id, +) +from tests import factories + +pytestmark = pytest.mark.integration + + +async def test_get_latest_run_id_returns_most_recent(db_session): + # Explicit started_at: now() is the (shared) txn timestamp, so ordering + # between two same-transaction rows would otherwise be ambiguous. + now = datetime.now(UTC) + await factories.make_simulation_run(db_session, started_at=now - timedelta(minutes=5)) + r2 = await factories.make_simulation_run(db_session, started_at=now) + latest = await get_latest_run_id(db_session) + assert latest == r2.id + + +async def test_record_pi_message_resolves_channel_and_writes_human_row(db_session): + run = await factories.make_simulation_run(db_session) + # A known channel with collab_private visibility should be picked up. + await factories.make_agent_channel( + db_session, run=run, channel_name="general", channel_id="C-GEN", + visibility="collab_private", + ) + msg = await record_pi_message( + db_session, run_id=run.id, channel_name="general", + content="please prioritize the kinase panel", sender_name="Dr Smoke (PI)", + ) + await db_session.flush() + + assert msg.is_bot is False # human/PI message + assert msg.agent_id is None # NULL sender_agent_id + assert msg.channel_id == "C-GEN" # resolved from agent_channels + assert msg.visibility == "collab_private" + assert msg.phase == "new_post" # top-level (no thread_ts) + assert msg.posted_at > 0 and msg.message_ts + + row = (await db_session.execute( + select(AgentMessage).where(AgentMessage.message_ts == msg.message_ts) + )).scalar_one() + assert row.content == "please prioritize the kinase panel" + assert row.sender_name == "Dr Smoke (PI)" + + +async def test_record_pi_message_reply_and_local_channel_fallback(db_session): + run = await factories.make_simulation_run(db_session) + # No agent_channels row for this name -> local: id, public visibility. + msg = await record_pi_message( + db_session, run_id=run.id, channel_name="drug-repurposing", + content="following up here", sender_name="PI", thread_ts="123.456", + ) + assert msg.channel_id == "local:drug-repurposing" + assert msg.visibility == "public" + assert msg.thread_ts == "123.456" + assert msg.phase == "thread_reply" # has a thread_ts + + +async def test_record_pi_dm_inbound_and_outbound(db_session): + run = await factories.make_simulation_run(db_session) + uid = uuid.uuid4() + inbound = await record_pi_dm( + db_session, run_id=run.id, agent_id="su", pi_user_id=web_pi_user_id(uid), + direction="inbound", content="always cc me on proposals", sender_name="PI", + ) + await record_pi_dm( + db_session, run_id=run.id, agent_id="su", pi_user_id=web_pi_user_id(uid), + direction="outbound", content="noted — will do", sender_name="SuBot", + ) + await db_session.flush() + + assert inbound.pi_user_id == f"local:{uid}" + rows = (await db_session.execute( + select(PiDmMessage).where(PiDmMessage.simulation_run_id == run.id) + .order_by(PiDmMessage.posted_at.asc()) + )).scalars().all() + assert [r.direction for r in rows] == ["inbound", "outbound"] + assert rows[0].content == "always cc me on proposals" + assert rows[1].agent_id == "su" + assert all(r.ts and r.posted_at > 0 for r in rows)