feat: control plane + risk governance layer for Sarthi#28
Conversation
…CE loop, MissionState, docs - Tool calling surface with ToolRegistry + 4 tools (pause retry, draft update, schedule checkin, flag churn) wired to HITL manager with correct string tiers - SSEHub event-type filtering: Subscription.Types gates Broadcast fan-out; 3 new SSE endpoints (/events/mission, /events/hitl, /events/session) - Slack consolidation: extended SocketModeClient for interactive events; removed Mockoon dead code from slack.py - ACE loop completed: _send_feedback_signal() now calls curator.update() - MissionState: added prepared_brief, pending_decisions, last_updated_by in Python dataclass + SQL queries (DB already done in migration 004) - Updated AGENTS.md, README.md, and all 5 .opencode/context files - Fixed hitl_queue_test.go: NewHandler now takes 2 args
- pause_payment_retry: Stripe POST /v1/subscriptions/{id} with
collection_method=send_invoice; httpx-based, MOCK_MODE guard
- draft_investor_update: loads MissionState + chat_completion()
for bounded (300 token) investor email draft
- schedule_customer_checkin: Slack chat_postMessage via
SlackClient to configurable channel; mock when no SLACK_BOT_TOKEN
- flag_churn_risk: updates churn_risk_users field via
get_mission_state() + update_mission_state(); decision journal log
- generate_prepared_brief(): 2-sentence LLM summary triggered
after MissionState write when prepared_brief is empty
- update_mission_state() accepts generate_brief: bool = True param
…nability, dashboard - Agent authority manifest: 5 agents with role/permissions/tool allowlists/ escalation tiers/mission field ownership. Machine-readable Pydantic model. - Alert lineage: AlertLineage schema attached to every GuardianMessage. pattern_id + source_metrics + mission_context + raise_timeline_risk are deterministic (code-only). suggested_tool_ids from HITLManager. owner_agent from manifest. - MissionState explainability: 3 new fields (last_update_reason, last_changed_fields, active_agent_roles) across Python dataclass, SQL migration 005, and Go read model. - Curator write verification: StrategyDelta Pydantic model for structured audit trail. update_strategy_confidence() returns ConfidenceUpdateResult + StrategyDelta tuple. PostgreSQL audit with JSONL fallback. - Dashboard operating layer: 2 new Go handlers (APICommandAlertLineage, APICommandOperatingLayer) with HTMX partials. 5 new Go tests. - All 24 Go web tests pass. Python assertions all pass.
last_changed_fields and active_agent_roles are Python lists but the DB columns are JSONB. asyncpg requires json.dumps() before passing list values to JSONB columns. Added import json and serialization at the parameter boundary.
Add lightweight control plane with agent registry, policy engine, and audit logging. Implement prompt risk and output risk scanners for investor-facing workflows. Force HITL approve on all external-facing outputs. Extend MissionState with policy_state explainability. New modules: - src/control_plane/ — registry, policy engine, audit logger - src/risk/ — prompt risk scanner (9 rules), output risk scanner (9 rules) - src/schemas/control_plane.py — PolicyDecision, RiskScanResult, AgentRegistration, AuditEvent Pydantic v2 models - infrastructure/migrations/004_control_plane_audit.sql - tests/unit/test_control_plane.py — 25 tests Modified: - authority_manifest.py — added allowed_models, external_facing, data_classification to all 5 agents - mission_state.py — added policy_state field with full DB roundtrip - draft_investor_update.py — risk-gated: prompt scan -> LLM -> output scan -> audit log, always requires_approval: True - schemas/__init__.py — exported new control plane models Preserves Sarthi's guardian architecture, deterministic watchlists, and bounded LLM usage. Zero regressions (799 unit tests pass).
📝 WalkthroughWalkthroughThis PR adds SSEHub event-type filtering, a Python ToolRegistry with HITL tier/pattern routing, a control-plane layer (registry, policy engine, audit logging), deterministic prompt/output risk scanners, Slack SocketMode consolidation with ACE feedback wiring, MissionState explainability fields with brief generation, new command-center HTMX endpoints/templates, and corresponding documentation. ChangesSarthi V4 Architecture Evolution
Estimated code review effort: 4 (Complex) | ~75 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Handler
participant SSEHub
Client->>Handler: GET /events/mission
Handler->>SSEHub: Subscribe(tenantID, "mission-update")
SSEHub-->>Handler: Subscription{ID, Channel, Types}
Handler->>Client: stream matching events
sequenceDiagram
participant Agent
participant PolicyEngine
participant ToolRegistry
participant AuditLogger
Agent->>PolicyEngine: evaluate(agent, requested_tool)
PolicyEngine-->>Agent: PolicyDecision
Agent->>ToolRegistry: get_tools_for_patterns(triggered_patterns)
ToolRegistry-->>Agent: matching ToolDef list
Agent->>AuditLogger: log_event(AuditEvent)
sequenceDiagram
participant SlackUser
participant SlackClient
participant SlackButtons
participant Curator
SlackUser->>SlackClient: interactive button click (Socket Mode)
SlackClient->>SlackButtons: route_slack_button()
SlackButtons->>Curator: update_strategy_confidence()
Curator-->>SlackButtons: ConfidenceUpdateResult, StrategyDelta
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 golangci-lint (2.12.2)level=error msg="[linters_context] typechecking error: pattern ./...: directory prefix . does not contain main module or its selected dependencies" Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces major architectural evolutions to the Sarthi platform, including a centralized SSEHub with event-type filtering, a global ToolRegistry mapped to HITL tiers, Slack SocketMode integration with an ACE feedback loop, and expanded MissionState fields for cognitive offloading and explainability. The review feedback identifies several critical issues: incorrect asynchronous calls and missing await keywords when logging audit events or connecting to Slack SocketMode; missing database connection parameters in the curator's audit logging; a policy engine bug where degraded health status completely blocks LLM models; strict string-matching mismatches in Go SSE subscriptions for mission, HITL, and session events; and a blocking synchronous LLM call inside an asynchronous brief generator that should be run in a separate thread.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| "reason": "Output risk scan blocked draft", | ||
| }, | ||
| ) | ||
| _AUDIT.log(audit_event) |
There was a problem hiding this comment.
The AuditLogger class in apps/ai/src/control_plane/audit.py defines the logging method as async def log_event(...). Calling _AUDIT.log(audit_event) will raise an AttributeError because log is not defined on AuditLogger. Additionally, since it is an asynchronous method, it must be awaited to execute.
| _AUDIT.log(audit_event) | |
| await _AUDIT.log_event(audit_event) |
| "output_scan": output_scan.model_dump(), | ||
| }, | ||
| ) | ||
| _AUDIT.log(audit_event) |
There was a problem hiding this comment.
The AuditLogger class in apps/ai/src/control_plane/audit.py defines the logging method as async def log_event(...). Calling _AUDIT.log(audit_event) will raise an AttributeError because log is not defined on AuditLogger. Additionally, since it is an asynchronous method, it must be awaited to execute.
| _AUDIT.log(audit_event) | |
| await _AUDIT.log_event(audit_event) |
| pass | ||
|
|
||
| self.socket_client.socket_mode_request_listeners.append(_process) | ||
| await self.socket_client.connect() |
There was a problem hiding this comment.
SocketModeClient imported from slack_sdk.socket_mode is the synchronous, thread-based client. Its connect() method is synchronous and returns None. Awaiting it will raise a TypeError: object NoneType can't be used in 'await' expression at runtime. To resolve this, remove the await keyword.
| await self.socket_client.connect() | |
| self.socket_client.connect() |
| import asyncio | ||
|
|
||
| async def _pg_write() -> None: | ||
| conn = await asyncpg.connect() |
There was a problem hiding this comment.
asyncpg.connect() is called without any connection parameters. This will fail to connect or attempt to connect to default local Postgres instances instead of using the configured database URL (get_database_url("iterateswarm")) used elsewhere in the application.
from src.config.database import get_database_url
conn = await asyncpg.connect(get_database_url("iterateswarm"))| if blocked_reason is not None: | ||
| allowed_models = [] |
There was a problem hiding this comment.
When an agent's health status is "degraded", the policy engine sets blocked_reason = "agent_health_degraded". However, any non-None blocked_reason causes allowed_models to be cleared (allowed_models = []). This completely blocks the agent from making LLM calls, contradicting the design that degraded health "forces review but not full block".
| if blocked_reason is not None: | |
| allowed_models = [] | |
| if blocked_reason is not None and blocked_reason not in ("agent_health_degraded",): | |
| allowed_models = [] |
| // APICommandMissionEvents is an SSE endpoint for mission state updates (event type: "mission-update"). | ||
| func (h *Handler) APICommandMissionEvents(c *fiber.Ctx) error { | ||
| tenantID := c.Query("tenant_id", "default") | ||
| sub := h.sseHub.Subscribe(tenantID, "mission-update") |
There was a problem hiding this comment.
The Go handler subscribes to the event type "mission-update". However, the system architecture documentation (chat-flow.md and README.md) states that the event types managed by SSEHub are "chat", "mission", "hitl", and "session". Because SSEHub.Broadcast performs strict string matching on the event type, any broadcast sent with the documented type "mission" will be silently dropped for subscribers listening to "mission-update".
| sub := h.sseHub.Subscribe(tenantID, "mission-update") | |
| sub := h.sseHub.Subscribe(tenantID, "mission") |
| // APICommandHITLEvents is an SSE endpoint for HITL approval events (event type: "hitl-item"). | ||
| func (h *Handler) APICommandHITLEvents(c *fiber.Ctx) error { | ||
| tenantID := c.Query("tenant_id", "default") | ||
| sub := h.sseHub.Subscribe(tenantID, "hitl-item") |
There was a problem hiding this comment.
The Go handler subscribes to the event type "hitl-item". However, the system architecture documentation (chat-flow.md and README.md) states that the event types managed by SSEHub are "chat", "mission", "hitl", and "session". Because SSEHub.Broadcast performs strict string matching on the event type, any broadcast sent with the documented type "hitl" will be silently dropped for subscribers listening to "hitl-item".
| sub := h.sseHub.Subscribe(tenantID, "hitl-item") | |
| sub := h.sseHub.Subscribe(tenantID, "hitl") |
| // APICommandSessionEvents is an SSE endpoint for agent message events (event type: "agent-message"). | ||
| func (h *Handler) APICommandSessionEvents(c *fiber.Ctx) error { | ||
| tenantID := c.Query("tenant_id", "default") | ||
| sub := h.sseHub.Subscribe(tenantID, "agent-message") |
There was a problem hiding this comment.
The Go handler subscribes to the event type "agent-message". However, the system architecture documentation (chat-flow.md and README.md) states that the event types managed by SSEHub are "chat", "mission", "hitl", and "session". Because SSEHub.Broadcast performs strict string matching on the event type, any broadcast sent with the documented type "session" will be silently dropped for subscribers listening to "agent-message".
| sub := h.sseHub.Subscribe(tenantID, "agent-message") | |
| sub := h.sseHub.Subscribe(tenantID, "session") |
| brief = chat_completion( | ||
| messages=[ | ||
| {"role": "system", "content": "You are a concise business briefing assistant. Output exactly 2 sentences. No preamble."}, | ||
| {"role": "user", "content": prompt}, | ||
| ], | ||
| max_tokens=80, | ||
| temperature=0.3, | ||
| ) |
There was a problem hiding this comment.
chat_completion is a synchronous, blocking network call. Calling it directly inside the asynchronous function generate_prepared_brief will block the entire event loop for the duration of the LLM request (typically 1-3 seconds), preventing other concurrent tasks (like SSE streaming or other requests) from executing. To avoid blocking the event loop, run the synchronous call in a separate thread using asyncio.to_thread.
import asyncio
brief = await asyncio.to_thread(
chat_completion,
messages=[
{"role": "system", "content": "You are a concise business briefing assistant. Output exactly 2 sentences. No preamble."},
{"role": "user", "content": prompt},
],
max_tokens=80,
temperature=0.3,
)![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 18
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
apps/core/internal/web/handler.go (1)
1627-1763: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winMove
Unsubscribeinto the stream writer
SetBodyStreamWriterruns after the handler returns, so the deferredUnsubscribeclosessub.Channelbefore the loop starts. The first receive from that closed channel will end the SSE stream right after the initialconnectedevent. Keep the cleanup inside the writer for all three handlers.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/core/internal/web/handler.go` around lines 1627 - 1763, The SSE handlers APICommandMissionEvents, APICommandHITLEvents, and APICommandSessionEvents unsubscribe too early because the deferred h.sseHub.Unsubscribe runs before SetBodyStreamWriter starts, closing sub.Channel and ending the stream after the initial connected event. Move the Unsubscribe cleanup inside the SetBodyStreamWriter closure (e.g., defer it at the start of the writer) so it runs when the stream actually finishes, while keeping the current subscribe/read loop structure unchanged.
♻️ Duplicate comments (1)
apps/core/internal/db/migrations/005_mission_state_explainability.sql (1)
1-8: 🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick winMigration appears incomplete relative to
mission_state.py's new columns.This migration only creates
last_update_reason,last_changed_fields,active_agent_roles, butmission_state.py'sget_mission_state/update_mission_statealso read/writeprepared_brief,pending_decisions,last_updated_by, andpolicy_stateon the same table (see comment on that file). If those 4 columns aren't already present from an earlier, unprovided migration, this migration should be extended to add them.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/core/internal/db/migrations/005_mission_state_explainability.sql` around lines 1 - 8, The migration for mission_states is incomplete compared with the fields used by mission_state.py. Extend the existing migration to also add the missing columns referenced by get_mission_state and update_mission_state: prepared_brief, pending_decisions, last_updated_by, and policy_state, alongside the current explainability fields. Make sure the ALTER TABLE in this migration matches all columns that the model now reads and writes so the schema stays in sync.
🟡 Minor comments (8)
.opencode/context/templates/architecture-overview.md-43-47 (1)
43-47: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winAdd a blank line before the Go Core table.
Markdownlint flags tables that start immediately after a heading. One blank line here keeps the template lint-clean.
Suggested fix
### Go Core Layer + | Community | Key Files | Description |🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In @.opencode/context/templates/architecture-overview.md around lines 43 - 47, The markdown table under the Go Core Layer heading starts too immediately after the heading, which triggers markdownlint. Update the architecture overview template so there is a blank line between the “### Go Core Layer” heading and the table, keeping the surrounding section structure unchanged.Source: Linters/SAST tools
README.md-7-7 (1)
7-7: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winReplace the empty badge target.
(#)is a dead link and triggers markdownlint. Point the badge at a real section or drop the wrapper.Suggested fix
-[](#) +[]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@README.md` at line 7, The tests badge in the README uses a dead empty anchor target, which triggers markdownlint; update the badge link in the README so it points to a real section in the document or remove the link wrapper entirely. Use the existing badge markup around the Tests badge as the identifier to locate and fix it.Source: Linters/SAST tools
.opencode/context/standards/coding-standards.md-178-227 (1)
178-227: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winMirror the actual
ToolDefcontract here.This example still teaches a dict-based
tool_defmodule, but the new tool surface is centered onToolDef,TOOL_REGISTRY, andregister_tool(). Please update the standard so new tools are scaffolded the same way the code is registered.Suggested fix
-from typing import Any - -tool_def: dict[str, Any] = { - "name": "my_tool_name", - "description": "Human-readable description", - "hitl_tier": "review", - "trigger_patterns": ["FG-05", "BG-04"], -} +from typing import Any +from src.agents.tools import ToolDef + +tool_def = ToolDef( + name="my_tool_name", + description="Human-readable description", + hitl_tier="review", + fn=execute, + trigger_patterns=["FG-05", "BG-04"], +) @@ -from . import my_tool_module - -register_tool(ToolDef(**my_tool_module.tool_def, fn=my_tool_module.execute)) +from . import my_tool_module + +register_tool(my_tool_module.tool_def)As per coding guidelines, tools under
apps/ai/src/agents/tools/**/*.pyshould useToolDef,TOOL_REGISTRY, andregister_tool()with auto-import in__init__.py.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In @.opencode/context/standards/coding-standards.md around lines 178 - 227, The Tool Creation Pattern example is outdated because it still shows a dict-based tool_def contract instead of the actual ToolDef-based registration flow. Update the standard to use ToolDef, TOOL_REGISTRY, and register_tool() as the canonical pattern, and describe that tools should be auto-registered via import in __init__.py. Make sure the example and bullets reference the real tool symbols used in apps/ai/src/agents/tools, such as ToolDef, register_tool(), and TOOL_REGISTRY, so new tools are scaffolded consistently with the current codebase.Source: Coding guidelines
apps/ai/src/control_plane/registry.py-36-39 (1)
36-39: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winMake
AgentRegistrationimmutable or return copies
ControlPlaneRegistry.get()andlist_agents()hand out liveAgentRegistrationinstances, andAgentRegistrationis a mutableBaseModel. Callers can mutate fields in place and bypass the registry lock/copy-on-write path inset_health(), which breaks the thread-safety guarantee.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/src/control_plane/registry.py` around lines 36 - 39, `ControlPlaneRegistry.get()` and `list_agents()` are returning live `AgentRegistration` objects, so callers can mutate them outside the registry lock and bypass the `set_health()` copy-on-write flow. Update the `AgentRegistration` model to be immutable or change `get()`/`list_agents()` to return defensive copies, and ensure the `ControlPlaneRegistry` methods remain the only place that mutates stored registrations.apps/ai/src/control_plane/audit.py-53-58 (1)
53-58: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winNormalize missing
policy_decisionto SQL NULL
policy_decisioncurrently stores JSON null when absent, whiledetailsstores SQL NULL. That makespolicy_decision IS NULLmiss rows with no decision; passNonethrough here as well so both columns share the same unset semantics.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/src/control_plane/audit.py` around lines 53 - 58, The audit row builder in audit.py currently serializes a missing policy_decision as JSON null instead of SQL NULL. Update the logic around the event.policy_decision model_dump call so it passes None through when policy_decision is absent, matching the unset handling already used for event.details. Keep the change localized to the audit insert path so policy_decision and details use the same null semantics.apps/ai/src/agents/tools/draft_investor_update.py-73-78 (1)
73-78: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winRun the LLM call off the event loop.
chat_completionis synchronous, so calling it directly fromasync def executeblocks the event loop for the full model round-trip. Wrap it inawait asyncio.to_thread(...)or switch to an async helper.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/src/agents/tools/draft_investor_update.py` around lines 73 - 78, The synchronous chat_completion call inside execute is blocking the event loop during the model request. Update draft_investor_update.py so the draft generation in execute runs off the loop, ideally by wrapping chat_completion in await asyncio.to_thread(...) or by switching to an async helper, and keep the existing metrics_prompt, max_tokens, and temperature usage intact.apps/ai/src/agents/tools/pause_payment_retry.py-45-64 (1)
45-64: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winAdd an
Idempotency-Keyto this Stripe POST. This request mutates subscription billing behavior, and Stripe recommends idempotency keys for POSTs so retries don’t repeat the operation or its side effects.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/src/agents/tools/pause_payment_retry.py` around lines 45 - 64, The Stripe subscription update in pause_payment_retry.py is a mutating POST and currently has no idempotency protection. Update the AsyncClient.post call in the pause_payment_retry flow to send an Idempotency-Key header, using a stable unique value for the operation (for example derived from subscription_id and tenant_id or an explicit request token) so retries on the same action do not repeat side effects.apps/core/internal/web/templates/partials/command_operating_layer.html-58-61 (1)
58-61: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winRender active agent roles as individual badges.
active_agent_rolesis stored as JSONB and cast to text in the handler, so this badge will show the raw array string (brackets/quotes) instead of a readable list. Decode it to[]stringand range over the items here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/core/internal/web/templates/partials/command_operating_layer.html` around lines 58 - 61, The ActiveAgentRoles display currently renders the raw JSON array string instead of separate readable badges. Update the command_operating_layer template to range over the decoded roles rather than printing .ActiveAgentRoles directly, and make sure the handler or template data exposes active_agent_roles as a []string value. Use the existing .ActiveAgentRoles field and the badge markup in this template to render one badge per role.
🧹 Nitpick comments (8)
apps/core/internal/web/handler.go (1)
1627-1763: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoffConsolidate the four near-identical SSE handlers.
APICommandChatEvents,APICommandMissionEvents,APICommandHITLEvents, andAPICommandSessionEventsdiffer only in the subscribed event type and theconnectedmessage text. Extract a shared helper to remove the duplication and keep the heartbeat/streaming logic in one place.♻️ Sketch of a shared helper
func (h *Handler) sseStream(c *fiber.Ctx, eventType, connectedText string) error { tenantID := c.Query("tenant_id", "default") sub := h.sseHub.Subscribe(tenantID, eventType) c.Set("Content-Type", "text/event-stream") c.Set("Cache-Control", "no-cache") c.Set("Connection", "keep-alive") done := c.Context().Done() c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { defer func() { recover() }() defer h.sseHub.Unsubscribe(tenantID, sub.ID) // cleanup inside writer (see prior comment) fmt.Fprintf(w, "event: connected\ndata: {\"status\":\"connected\",\"text\":%q}\n\n", connectedText) w.Flush() heartbeat := time.NewTicker(30 * time.Second) defer heartbeat.Stop() for { select { case <-heartbeat.C: if _, err := fmt.Fprintf(w, "event: heartbeat\ndata: {}\n\n"); err != nil { return } w.Flush() case msgBytes, ok := <-sub.Channel: if !ok { return } if _, err := fmt.Fprintf(w, "%s", msgBytes); err != nil { return } w.Flush() case <-done: return } } }) return nil }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/core/internal/web/handler.go` around lines 1627 - 1763, The SSE handlers are duplicated across APICommandMissionEvents, APICommandHITLEvents, and APICommandSessionEvents (matching the existing APICommandChatEvents pattern), with only the subscribed event type and connected text changing. Extract the shared heartbeat/streaming logic into a helper like sseStream on Handler, pass in eventType and connectedText, and update each APICommand*Events method to delegate to it so all SSE behavior lives in one place.apps/ai/src/integrations/slack_buttons.py (1)
98-128: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
except (ImportError, Exception)is redundant and silently swallows every failure.
ImportErroris a subclass ofException, so the tuple collapses to a bareexcept Exception. Across all three ACE steps this discards errors with no logging, making feedback-loop failures invisible in production. Consider logging at least atdebug/warningso silent breakage (e.g. the guardian/curator issues elsewhere in this stack) is observable.♻️ Example for one step
try: from src.agents.cofounder.reflector import score_from_button score_from_button(alert_id, response_type, score) - except (ImportError, Exception): - pass + except Exception: + logger.debug("Reflector scoring step failed", exc_info=True)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/src/integrations/slack_buttons.py` around lines 98 - 128, The three ACE update blocks in slack_buttons.py are swallowing all failures because `except (ImportError, Exception)` is effectively a bare `except Exception` and ignores errors silently. Update the `score_from_button`, `update_strategy_confidence`, and `Curator.update` try/except blocks to catch only the needed exception type(s), and add logging in each handler so failures are observable instead of being hidden.apps/ai/src/agents/cofounder/curator.py (1)
387-400: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueAvoid running
CREATE TABLEDDL on the audit write hot path.Issuing
CREATE TABLE IF NOT EXISTSon every audit write adds a round-trip per call and managesstrategy_deltasoutside the migration system (this PR already introduces migration004_control_plane_audit.sql). Prefer declaringstrategy_deltasas a migration and keeping this function to inserts only.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/src/agents/cofounder/curator.py` around lines 387 - 400, The audit write path in curator.py is creating the strategy_deltas table inline instead of relying on the migration system. Remove the CREATE TABLE IF NOT EXISTS block from the audit insert flow in the function that writes strategy deltas, and ensure strategy_deltas is defined only in migration 004_control_plane_audit.sql so this path only performs inserts.apps/ai/infrastructure/migrations/004_control_plane_audit.sql (1)
3-16: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winConsider CHECK constraints for enum-like columns.
approval_stateandoutcomeare documented (viaCOMMENT) as constrained value sets but aren't enforced with aCHECKconstraint, so invalid values could be silently persisted if the Python layer's validation is ever bypassed.♻️ Optional hardening
approval_state VARCHAR(20), outcome VARCHAR(20) NOT NULL, + CONSTRAINT chk_audit_log_approval_state CHECK (approval_state IN ('auto','review','approve','blocked')), + CONSTRAINT chk_audit_log_outcome CHECK (outcome IN ('completed','blocked','failed')),🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/infrastructure/migrations/004_control_plane_audit.sql` around lines 3 - 16, The audit_log table defines enum-like fields without enforcing their allowed values, so add CHECK constraints for approval_state and outcome in the migration. Update the CREATE TABLE in the audit_log definition to validate the documented value sets directly at the database level, keeping the constraints aligned with the existing comments and schema intent.apps/ai/tests/unit/test_control_plane.py (1)
234-317: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a test for the degraded-health branch of
PolicyEngine.evaluate.All other branches (
unhealthy,restricted,external_facing,disallowed tool) are covered, buthealth_status == "degraded"is not. Adding this test would have caught theallowed_model_classeswipe issue flagged inapps/ai/src/control_plane/policy.py.✅ Suggested test
def test_degraded_agent_requires_review_but_keeps_models(self): engine = PolicyEngine() reg = AgentRegistration( agent_name="DegradedAgent", role="ops", domain="ops", allowed_tools=["flag_churn_risk_customer"], allowed_models=["gpt-4o-mini"], escalation_tier="auto", external_facing=False, data_classification="internal", health_status="degraded", ) decision = engine.evaluate(reg, requested_tool="flag_churn_risk_customer") assert decision.requires_human_approval is True assert "flag_churn_risk_customer" in decision.approved_tools assert "gpt-4o-mini" in decision.allowed_model_classes🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/tests/unit/test_control_plane.py` around lines 234 - 317, Add a unit test for the degraded-health path in TestPolicyEngine to cover PolicyEngine.evaluate when AgentRegistration.health_status is "degraded". Create a case like the existing policy tests, call evaluate with a permitted tool, and assert it requires human approval while still preserving approved_tools and allowed_model_classes (for example, the requested tool and "gpt-4o-mini" should remain allowed). Use the existing PolicyEngine and AgentRegistration test patterns so the branch in apps/ai/src/control_plane/policy.py is exercised.apps/ai/src/schemas/control_plane.py (1)
39-53: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winUse
Literalfordata_classificationinstead ofstr.The docstring enumerates exact valid values (
"internal","external_investor","external_customer","restricted"), but the field accepts any string. The same looseness exists onAgentRegistration.data_classification(Line 104). GivenPolicyEnginedoes case-insensitive set membership checks against this value, a typo (e.g."Internal ") would silently bypass intended gating instead of failing validation at the schema boundary.♻️ Proposed fix
+DataClassification = Literal["internal", "external_investor", "external_customer", "restricted"] + class PolicyDecision(BaseModel): ... - data_classification: str + data_classification: DataClassification🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/src/schemas/control_plane.py` around lines 39 - 53, `PolicyDecision.data_classification` is too permissive: replace the plain string type with a `Literal` of the four documented values so invalid classifications fail validation at the schema boundary. Apply the same tightening to `AgentRegistration.data_classification`, and keep the exact allowed values aligned with the `PolicyEngine` checks so typos or casing/whitespace variants are rejected instead of flowing through as valid input.apps/ai/src/control_plane/audit.py (1)
20-78: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winUse a shared pool when
connis not provided.asyncpg.connect()/close()on every call adds avoidable overhead; reuse a long-lived pool here and add a connect timeout to bound failures.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/src/control_plane/audit.py` around lines 20 - 78, The log_event method in audit.py currently opens and closes a fresh asyncpg connection on each call when conn is None, which adds avoidable overhead and lacks bounded connect behavior. Update log_event to use a shared asyncpg pool for the own_conn path instead of asyncpg.connect()/close(), and add a connect timeout to the pool acquisition or connection setup so failures are bounded. Keep the existing conn reuse path intact and make sure the change is localized around log_event and any pool helper used by Audit.apps/ai/src/agents/tools/draft_investor_update.py (1)
68-68: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winInconsistent response key naming across blocked branches.
Prompt-block returns
"scan_result"while output-block returns"risk_scan"for equivalent data, complicating downstream consumers that need to branch on either outcome.Also applies to: 100-100
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/ai/src/agents/tools/draft_investor_update.py` at line 68, The blocked-branch response keys are inconsistent for the same prompt-scan data, since the prompt-block in draft_investor_update uses scan_result while the output-block uses risk_scan. Update the relevant return paths in the draft_investor_update tool so both branches use the same key name, and align any downstream references to that shared symbol to avoid special-casing by branch.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/ai/src/agents/cofounder/curator.py`:
- Around line 420-433: The `_write_audit_log` fallback path is not catching
runtime DB write failures, so JSONL never runs when PostgreSQL write attempts
fail from a sync caller. Update `_write_audit_log` to ensure the PostgreSQL
write block (including the `asyncio.get_running_loop()` / `ThreadPoolExecutor`
path) catches connection/runtime exceptions and falls back to
`_file_audit_write(delta)`, while keeping only `ImportError` as the
SQLite/PostgreSQL-unavailable case. Also fix the `pool.submit(asyncio.run,
_pg_write)` call in `_write_audit_log` to run the coroutine object with
`asyncio.run(_pg_write())`.
In `@apps/ai/src/agents/tools/draft_investor_update.py`:
- Around line 26-28: The control-plane objects are created but not used, so
`execute()` can call `chat_completion()` without any registry lookup or policy
gate. Update `DraftInvestorUpdateTool.execute()` to consult `_REGISTRY` and
`_POLICY` before generating the draft, using the existing
`ControlPlaneRegistry`, `PolicyEngine`, and `_AUDIT` flow to validate
access/classification/model approval and short-circuit when denied. Keep the
policy decision close to the `chat_completion()` call so the investor-facing
draft only proceeds after the control-plane check passes.
- Around line 63-71: The audit logging in draft_investor_update is using the
wrong API and misses the blocked prompt-scan path, so update every _AUDIT call
to use AuditLogger.log_event(...) instead of _AUDIT.log(...) and ensure the
prompt_scan.recommended_action == "block" branch also creates and awaits an
AuditEvent before returning. Use the existing draft_investor_update flow and the
later success/output-block audit paths as references so all branches
consistently persist audit rows.
In `@apps/ai/src/agents/tools/flag_churn_risk.py`:
- Around line 36-41: The `flag_churn_risk` flow is using raw substring checks on
`state.churn_risk_users` and a non-atomic read-modify-write update. Update the
`flag_churn_risk` tool to treat `churn_risk_users` as a parsed list of segment
IDs (not a comma-joined string) so membership is checked by exact element match,
and change the `get_mission_state`/`update_mission_state` path to an atomic
append or locked update so concurrent calls cannot overwrite each other’s
additions. Use the existing `flag_churn_risk` function and
`update_mission_state` call site as the main places to adjust.
In `@apps/ai/src/agents/tools/schedule_customer_checkin.py`:
- Around line 46-57: The `execute` path in `schedule_customer_checkin` is
blocking the event loop because `SlackClient.client` is a synchronous
`WebClient` and `chat_postMessage()` is called directly. Update the `execute`
method to use a non-blocking approach by either wrapping the `chat_postMessage`
call with `asyncio.to_thread(...)` or switching this flow to `AsyncWebClient`,
and keep the rest of the response assembly (including `resp.get("ts", "")`)
unchanged.
In `@apps/ai/src/control_plane/policy.py`:
- Around line 70-92: The degraded-health branch in policy evaluation is
incorrectly clearing allowed model classes because blocked_reason is used for
both hard blocks and review-only states. In the policy logic around the
agent.external_facing, agent.health_status, and requested_tool checks, separate
hard-block state from review-only approval so the degraded path still preserves
allowed_models while setting requires_human_approval. Update the final Step 6
behavior in the policy function to only empty allowed_models for true deny
reasons, and add coverage for the degraded-health case with an allowed tool
request to verify models remain available.
In `@apps/ai/src/integrations/slack_client.py`:
- Around line 65-69: The Slack Socket Mode startup in SlackClient.connect is
awaiting a synchronous SocketModeClient.connect() call, which will fail and stop
listener initialization. Update the connect path to use the sync connect() call
on self.socket_client, or if async behavior is required, switch the import and
type usage to slack_sdk.socket_mode.aiohttp.SocketModeClient so the awaited call
matches the client implementation. Keep the listener registration and
_socket_listener_running update in the same flow after a successful connection.
- Around line 50-63: The Socket Mode listener in slack_client is acknowledging
requests incorrectly and re-parsing an already-parsed payload. Update the
interactive branch to pass req.payload directly into route_slack_button instead
of calling json.loads, and replace both req.ack() usages in the listener and
exception handler with
client.send_socket_mode_response(SocketModeResponse(envelope_id=req.envelope_id))
so the request is properly acknowledged.
In `@apps/ai/src/risk/output_risk.py`:
- Around line 17-45: The OC009 draft/pending-review check is being applied in
the wrong direction, causing a missing-disclaimer flag to be added when the
disclaimer is actually present. Update the logic around _MISSING_APPROVAL and
_ALL_OUTPUT_PATTERNS in output_risk.py so the shared pattern list is not used to
generate the “missing disclaimer” risk in the main scan, and keep the true
missing-case handling in the separate approval check. Ensure the main risk scan
only flags actual absence of the disclaimer, not matches of the disclaimer text
itself.
- Around line 17-45: The output risk patterns currently use capturing groups
that cause re.findall in the scanning flow to return tuples instead of the full
matched text. Update the matcher in output_risk.py so the scan path for
_ALL_OUTPUT_PATTERNS, including OC001, OC002, OC004, OC005, OC006, OC007, and
OC008, uses re.search and takes group(0) (or otherwise extracts the full match)
before building RiskFlag. Make sure the matched_text value propagated into
output_scan.model_dump() and the audit log stays the actual substring, not a
tuple repr.
In `@apps/ai/src/risk/prompt_risk.py`:
- Around line 17-41: The pattern definitions in prompt_risk.py are still using
capture groups that make re.findall return tuples, which breaks clean matched
text for RiskFlag.matched_text. Update the affected entries in
_INVESTOR_SENSITIVE_KEYWORDS, _EXTERNAL_SEND_PATTERNS, and any similar patterns
like R001 in _RESTRICTED_PATTERNS to use non-capturing groups or otherwise avoid
nested captures, matching the same fix used in output_risk.py. Keep the
identifiers (I001, I002, E001, E002, R001) and verify that the downstream
RiskFlag/AuditEvent path now receives plain string matches instead of tuple
reprs.
In `@apps/ai/src/schemas/control_plane.py`:
- Around line 16-21: The RiskFlag matched_text field is being persisted through
prompt_scan.model_dump() and output_scan.model_dump() into AuditEvent.details,
so redact it before it reaches AuditLogger serialization. Update the
draft_investor_update scan-result handling to hash or mask matched_text on the
RiskFlag model or immediately before building AuditEvent.details, while keeping
the rest of the RiskFlag fields intact.
In `@apps/ai/src/schemas/guardian.py`:
- Around line 32-47: `GuardianMessage.lineage` is now required but several
existing `GuardianMessage(...)` call sites still omit it, so update the
`GuardianMessage` model and all constructors in the affected tests/callers to
either provide a valid `AlertLineage` or make `lineage` optional with a safe
default. Use the `AlertLineage` schema and the `GuardianMessage` constructor
sites in the unit and agentic tests to ensure validation no longer fails for
existing usages.
In `@apps/ai/src/session/brief_generator.py`:
- Around line 29-38: The async flow in generate_prepared_brief currently calls
the synchronous chat_completion() directly, which can block the shared event
loop. Update this call site to run chat_completion() off the loop using await
asyncio.to_thread(...) or replace it with an async LLM client, keeping the rest
of the brief generation logic unchanged.
In `@apps/ai/src/session/mission_state.py`:
- Around line 196-198: The attribution fields in mission state are being updated
under the unrelated generate_brief flag, which causes caller-provided
update_reason/changed_fields to be dropped or existing state values to be
overwritten with None. In the mission_state update logic, move the
last_update_reason and last_changed_fields assignment out of the generate_brief
branch and only set them when the corresponding function arguments are actually
provided, preserving any values already on state; use the surrounding update
routine and the state.last_update_reason/state.last_changed_fields fields to
locate the fix.
- Around line 277-278: `pending_decisions` is being passed to an asyncpg jsonb
parameter as a raw Python list in the mission state save path, unlike the other
JSONB fields. Update the serialization in the mission-state persistence code
that uses `state.prepared_brief` and `state.pending_decisions` so
`pending_decisions` is converted with `json.dumps(state.pending_decisions) if
state.pending_decisions else '[]'` before binding, matching the handling used
for the other jsonb values.
- Around line 127-130: Add a schema migration for policy_state in the mission
state model, since it is referenced by get_mission_state and
update_mission_state but does not yet have a matching schema change. Update the
migration set that already covers the other mission_state fields so policy_state
is included in the database schema and any relevant drift-fix SQL, using the
mission_state data model as the locator for the missing field. Ensure the
migration is applied consistently with the existing mission state schema
changes.
In `@apps/core/internal/web/sse_hub.go`:
- Around line 55-65: The Unsubscribe method on SSEHub removes a subID but leaves
an empty tenant bucket behind in the subscribers map. Update Unsubscribe to
delete the tenantID entry from h.subscribers when the inner subs map becomes
empty after delete(subs, subID), using the existing SSEHub and Unsubscribe
symbols to locate the cleanup logic. Keep the channel close and subscription
removal behavior the same, but add the empty-map check before returning so
tenant buckets do not accumulate.
---
Outside diff comments:
In `@apps/core/internal/web/handler.go`:
- Around line 1627-1763: The SSE handlers APICommandMissionEvents,
APICommandHITLEvents, and APICommandSessionEvents unsubscribe too early because
the deferred h.sseHub.Unsubscribe runs before SetBodyStreamWriter starts,
closing sub.Channel and ending the stream after the initial connected event.
Move the Unsubscribe cleanup inside the SetBodyStreamWriter closure (e.g., defer
it at the start of the writer) so it runs when the stream actually finishes,
while keeping the current subscribe/read loop structure unchanged.
---
Minor comments:
In @.opencode/context/standards/coding-standards.md:
- Around line 178-227: The Tool Creation Pattern example is outdated because it
still shows a dict-based tool_def contract instead of the actual ToolDef-based
registration flow. Update the standard to use ToolDef, TOOL_REGISTRY, and
register_tool() as the canonical pattern, and describe that tools should be
auto-registered via import in __init__.py. Make sure the example and bullets
reference the real tool symbols used in apps/ai/src/agents/tools, such as
ToolDef, register_tool(), and TOOL_REGISTRY, so new tools are scaffolded
consistently with the current codebase.
In @.opencode/context/templates/architecture-overview.md:
- Around line 43-47: The markdown table under the Go Core Layer heading starts
too immediately after the heading, which triggers markdownlint. Update the
architecture overview template so there is a blank line between the “### Go Core
Layer” heading and the table, keeping the surrounding section structure
unchanged.
In `@apps/ai/src/agents/tools/draft_investor_update.py`:
- Around line 73-78: The synchronous chat_completion call inside execute is
blocking the event loop during the model request. Update
draft_investor_update.py so the draft generation in execute runs off the loop,
ideally by wrapping chat_completion in await asyncio.to_thread(...) or by
switching to an async helper, and keep the existing metrics_prompt, max_tokens,
and temperature usage intact.
In `@apps/ai/src/agents/tools/pause_payment_retry.py`:
- Around line 45-64: The Stripe subscription update in pause_payment_retry.py is
a mutating POST and currently has no idempotency protection. Update the
AsyncClient.post call in the pause_payment_retry flow to send an Idempotency-Key
header, using a stable unique value for the operation (for example derived from
subscription_id and tenant_id or an explicit request token) so retries on the
same action do not repeat side effects.
In `@apps/ai/src/control_plane/audit.py`:
- Around line 53-58: The audit row builder in audit.py currently serializes a
missing policy_decision as JSON null instead of SQL NULL. Update the logic
around the event.policy_decision model_dump call so it passes None through when
policy_decision is absent, matching the unset handling already used for
event.details. Keep the change localized to the audit insert path so
policy_decision and details use the same null semantics.
In `@apps/ai/src/control_plane/registry.py`:
- Around line 36-39: `ControlPlaneRegistry.get()` and `list_agents()` are
returning live `AgentRegistration` objects, so callers can mutate them outside
the registry lock and bypass the `set_health()` copy-on-write flow. Update the
`AgentRegistration` model to be immutable or change `get()`/`list_agents()` to
return defensive copies, and ensure the `ControlPlaneRegistry` methods remain
the only place that mutates stored registrations.
In `@apps/core/internal/web/templates/partials/command_operating_layer.html`:
- Around line 58-61: The ActiveAgentRoles display currently renders the raw JSON
array string instead of separate readable badges. Update the
command_operating_layer template to range over the decoded roles rather than
printing .ActiveAgentRoles directly, and make sure the handler or template data
exposes active_agent_roles as a []string value. Use the existing
.ActiveAgentRoles field and the badge markup in this template to render one
badge per role.
In `@README.md`:
- Line 7: The tests badge in the README uses a dead empty anchor target, which
triggers markdownlint; update the badge link in the README so it points to a
real section in the document or remove the link wrapper entirely. Use the
existing badge markup around the Tests badge as the identifier to locate and fix
it.
---
Duplicate comments:
In `@apps/core/internal/db/migrations/005_mission_state_explainability.sql`:
- Around line 1-8: The migration for mission_states is incomplete compared with
the fields used by mission_state.py. Extend the existing migration to also add
the missing columns referenced by get_mission_state and update_mission_state:
prepared_brief, pending_decisions, last_updated_by, and policy_state, alongside
the current explainability fields. Make sure the ALTER TABLE in this migration
matches all columns that the model now reads and writes so the schema stays in
sync.
---
Nitpick comments:
In `@apps/ai/infrastructure/migrations/004_control_plane_audit.sql`:
- Around line 3-16: The audit_log table defines enum-like fields without
enforcing their allowed values, so add CHECK constraints for approval_state and
outcome in the migration. Update the CREATE TABLE in the audit_log definition to
validate the documented value sets directly at the database level, keeping the
constraints aligned with the existing comments and schema intent.
In `@apps/ai/src/agents/cofounder/curator.py`:
- Around line 387-400: The audit write path in curator.py is creating the
strategy_deltas table inline instead of relying on the migration system. Remove
the CREATE TABLE IF NOT EXISTS block from the audit insert flow in the function
that writes strategy deltas, and ensure strategy_deltas is defined only in
migration 004_control_plane_audit.sql so this path only performs inserts.
In `@apps/ai/src/agents/tools/draft_investor_update.py`:
- Line 68: The blocked-branch response keys are inconsistent for the same
prompt-scan data, since the prompt-block in draft_investor_update uses
scan_result while the output-block uses risk_scan. Update the relevant return
paths in the draft_investor_update tool so both branches use the same key name,
and align any downstream references to that shared symbol to avoid
special-casing by branch.
In `@apps/ai/src/control_plane/audit.py`:
- Around line 20-78: The log_event method in audit.py currently opens and closes
a fresh asyncpg connection on each call when conn is None, which adds avoidable
overhead and lacks bounded connect behavior. Update log_event to use a shared
asyncpg pool for the own_conn path instead of asyncpg.connect()/close(), and add
a connect timeout to the pool acquisition or connection setup so failures are
bounded. Keep the existing conn reuse path intact and make sure the change is
localized around log_event and any pool helper used by Audit.
In `@apps/ai/src/integrations/slack_buttons.py`:
- Around line 98-128: The three ACE update blocks in slack_buttons.py are
swallowing all failures because `except (ImportError, Exception)` is effectively
a bare `except Exception` and ignores errors silently. Update the
`score_from_button`, `update_strategy_confidence`, and `Curator.update`
try/except blocks to catch only the needed exception type(s), and add logging in
each handler so failures are observable instead of being hidden.
In `@apps/ai/src/schemas/control_plane.py`:
- Around line 39-53: `PolicyDecision.data_classification` is too permissive:
replace the plain string type with a `Literal` of the four documented values so
invalid classifications fail validation at the schema boundary. Apply the same
tightening to `AgentRegistration.data_classification`, and keep the exact
allowed values aligned with the `PolicyEngine` checks so typos or
casing/whitespace variants are rejected instead of flowing through as valid
input.
In `@apps/ai/tests/unit/test_control_plane.py`:
- Around line 234-317: Add a unit test for the degraded-health path in
TestPolicyEngine to cover PolicyEngine.evaluate when
AgentRegistration.health_status is "degraded". Create a case like the existing
policy tests, call evaluate with a permitted tool, and assert it requires human
approval while still preserving approved_tools and allowed_model_classes (for
example, the requested tool and "gpt-4o-mini" should remain allowed). Use the
existing PolicyEngine and AgentRegistration test patterns so the branch in
apps/ai/src/control_plane/policy.py is exercised.
In `@apps/core/internal/web/handler.go`:
- Around line 1627-1763: The SSE handlers are duplicated across
APICommandMissionEvents, APICommandHITLEvents, and APICommandSessionEvents
(matching the existing APICommandChatEvents pattern), with only the subscribed
event type and connected text changing. Extract the shared heartbeat/streaming
logic into a helper like sseStream on Handler, pass in eventType and
connectedText, and update each APICommand*Events method to delegate to it so all
SSE behavior lives in one place.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: f057f382-7090-4f1b-ab52-c5a399bd5f84
⛔ Files ignored due to path filters (1)
apps/ai/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (40)
.opencode/context/adr/001-sarthi-v4-architecture-evolution.md.opencode/context/domain/architecture.md.opencode/context/domain/chat-flow.md.opencode/context/standards/coding-standards.md.opencode/context/templates/architecture-overview.mdAGENTS.mdREADME.mdapps/ai/infrastructure/migrations/004_control_plane_audit.sqlapps/ai/pyproject.tomlapps/ai/src/agents/authority_manifest.pyapps/ai/src/agents/cofounder/curator.pyapps/ai/src/agents/tools/__init__.pyapps/ai/src/agents/tools/draft_investor_update.pyapps/ai/src/agents/tools/flag_churn_risk.pyapps/ai/src/agents/tools/pause_payment_retry.pyapps/ai/src/agents/tools/schedule_customer_checkin.pyapps/ai/src/control_plane/__init__.pyapps/ai/src/control_plane/audit.pyapps/ai/src/control_plane/policy.pyapps/ai/src/control_plane/registry.pyapps/ai/src/hitl/manager.pyapps/ai/src/integrations/slack.pyapps/ai/src/integrations/slack_buttons.pyapps/ai/src/integrations/slack_client.pyapps/ai/src/risk/__init__.pyapps/ai/src/risk/output_risk.pyapps/ai/src/risk/prompt_risk.pyapps/ai/src/schemas/__init__.pyapps/ai/src/schemas/control_plane.pyapps/ai/src/schemas/guardian.pyapps/ai/src/session/brief_generator.pyapps/ai/src/session/mission_state.pyapps/ai/tests/unit/test_control_plane.pyapps/core/hitl_queue_test.goapps/core/internal/db/migrations/005_mission_state_explainability.sqlapps/core/internal/web/command_center_test.goapps/core/internal/web/handler.goapps/core/internal/web/sse_hub.goapps/core/internal/web/templates/partials/command_alert_lineage.htmlapps/core/internal/web/templates/partials/command_operating_layer.html
| try: | ||
| loop = asyncio.get_running_loop() | ||
| if loop.is_running(): | ||
| import concurrent.futures | ||
| with concurrent.futures.ThreadPoolExecutor() as pool: | ||
| pool.submit(asyncio.run, _pg_write).result(timeout=5) | ||
| else: | ||
| asyncio.run(_pg_write()) | ||
| except Exception: | ||
| raise # re-raise to trigger fallback | ||
|
|
||
| except ImportError: | ||
| # PostgreSQL not available — write to JSONL file | ||
| _file_audit_write(delta) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the file and relevant symbols.
git ls-files 'apps/ai/src/agents/cofounder/curator.py'
rg -n "_write_audit_log|_file_audit_write|asyncio.get_running_loop|asyncio.run\(|ThreadPoolExecutor|_send_feedback_signal|update_strategy_confidence" apps/ai/src/agents/cofounder/curator.py
# Show the relevant section with line numbers.
sed -n '360,470p' apps/ai/src/agents/cofounder/curator.pyRepository: Aparnap2/Track_Guard
Length of output: 3802
🏁 Script executed:
python3 - <<'PY'
import asyncio
print("probe1: get_running_loop outside running loop")
try:
asyncio.get_running_loop()
except Exception as e:
print(type(e).__name__, str(e))
async def coro():
return 123
print("probe2: asyncio.run on coroutine object")
try:
print(asyncio.run(coro()))
except Exception as e:
print(type(e).__name__, str(e))
print("probe3: asyncio.run on coroutine function")
try:
print(asyncio.run(coro))
except Exception as e:
print(type(e).__name__, str(e))
PYRepository: Aparnap2/Track_Guard
Length of output: 400
Restore the JSONL fallback in _write_audit_log
asyncio.get_running_loop()raises in the sync caller, so the DB-write failure escapes toupdate_strategy_confidencebefore_file_audit_write()can run.- The outer fallback only handles
ImportError; connection/runtime failures never reach JSONL. pool.submit(asyncio.run, _pg_write)passes the function, not the coroutine object; useasyncio.run(_pg_write()).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/ai/src/agents/cofounder/curator.py` around lines 420 - 433, The
`_write_audit_log` fallback path is not catching runtime DB write failures, so
JSONL never runs when PostgreSQL write attempts fail from a sync caller. Update
`_write_audit_log` to ensure the PostgreSQL write block (including the
`asyncio.get_running_loop()` / `ThreadPoolExecutor` path) catches
connection/runtime exceptions and falls back to `_file_audit_write(delta)`,
while keeping only `ImportError` as the SQLite/PostgreSQL-unavailable case. Also
fix the `pool.submit(asyncio.run, _pg_write)` call in `_write_audit_log` to run
the coroutine object with `asyncio.run(_pg_write())`.
| _REGISTRY = ControlPlaneRegistry() | ||
| _POLICY = PolicyEngine() | ||
| _AUDIT = AuditLogger() |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== target file ==\n'
sed -n '1,220p' apps/ai/src/agents/tools/draft_investor_update.py
printf '\n== locate symbols ==\n'
rg -n "class AuditLogger|def log_event|def log\(|class PolicyEngine|class ControlPlaneRegistry|def chat_completion|async def chat_completion|def scan_output|log_event\(" apps/ai/src -S
printf '\n== relevant surrounding files (first hits) ==\n'
for f in \
$(rg -l "class AuditLogger|def log_event|class PolicyEngine|class ControlPlaneRegistry|def chat_completion|async def chat_completion|def scan_output" apps/ai/src -S | head -n 20); do
echo "---- $f ----"
sed -n '1,240p' "$f"
doneRepository: Aparnap2/Track_Guard
Length of output: 26677
Wire the control-plane policy check into execute() ControlPlaneRegistry() and PolicyEngine() are instantiated but never used, so this tool can reach chat_completion() without any registry lookup or policy decision. That bypasses the intended access/classification/model gate for an investor-facing draft.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/ai/src/agents/tools/draft_investor_update.py` around lines 26 - 28, The
control-plane objects are created but not used, so `execute()` can call
`chat_completion()` without any registry lookup or policy gate. Update
`DraftInvestorUpdateTool.execute()` to consult `_REGISTRY` and `_POLICY` before
generating the draft, using the existing `ControlPlaneRegistry`, `PolicyEngine`,
and `_AUDIT` flow to validate access/classification/model approval and
short-circuit when denied. Keep the policy decision close to the
`chat_completion()` call so the investor-facing draft only proceeds after the
control-plane check passes.
| prompt_scan = scan_prompt(metrics_prompt, context="investor_update") | ||
| if prompt_scan.recommended_action == "block": | ||
| log.warning("draft_investor_update %s — prompt risk scan blocked", tenant_id) | ||
| return { | ||
| "error": "Prompt risk scan blocked", | ||
| "scan_result": prompt_scan.model_dump(), | ||
| "tenant_id": tenant_id, | ||
| "requires_approval": True, | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== File map ==\n'
git ls-files apps/ai/src/agents/tools/draft_investor_update.py apps/ai/src/control_plane/audit.py
printf '\n== Relevant outlines ==\n'
ast-grep outline apps/ai/src/agents/tools/draft_investor_update.py --view expanded || true
printf '\n---\n'
ast-grep outline apps/ai/src/control_plane/audit.py --view expanded || true
printf '\n== Targeted search for audit usage ==\n'
rg -n "_AUDIT|log_event|\.log\(" apps/ai/src/agents/tools/draft_investor_update.py apps/ai/src/control_plane/audit.py
printf '\n== Relevant slices from draft_investor_update.py ==\n'
sed -n '1,220p' apps/ai/src/agents/tools/draft_investor_update.py
printf '\n== Relevant slices from audit.py ==\n'
sed -n '1,220p' apps/ai/src/control_plane/audit.pyRepository: Aparnap2/Track_Guard
Length of output: 7851
Use await _AUDIT.log_event(...) on all audit paths (draft_investor_update.py:63-71, 97, 120).
AuditLoggeronly exposeslog_event(...); the current_AUDIT.log(...)calls will fail at runtime, so the success/output-block paths never persist audit rows.- The
prompt_scan.recommended_action == "block"branch returns before anyAuditEventis created, leaving the highest-risk path without an audit trail.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/ai/src/agents/tools/draft_investor_update.py` around lines 63 - 71, The
audit logging in draft_investor_update is using the wrong API and misses the
blocked prompt-scan path, so update every _AUDIT call to use
AuditLogger.log_event(...) instead of _AUDIT.log(...) and ensure the
prompt_scan.recommended_action == "block" branch also creates and awaits an
AuditEvent before returning. Use the existing draft_investor_update flow and the
later success/output-block audit paths as references so all branches
consistently persist audit rows.
| state = await get_mission_state(tenant_id) | ||
| existing = state.churn_risk_users or "" | ||
| if segment_id not in existing: | ||
| state.churn_risk_users = (existing + "," + segment_id).strip(",") | ||
| state.last_updated_by = "flag_churn_risk_tool" | ||
| await update_mission_state(state) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Substring membership check and read-modify-write race on churn_risk_users.
segment_id not in existing does raw substring matching against a comma-joined string, so e.g. "seg1" would be considered present if "seg12" already exists (false negative) or vice versa. Separately, get_mission_state → update_mission_state is a non-atomic read-modify-write: two concurrent calls for the same tenant with different segment_ids can race, with the last write silently dropping the other's flagged segment.
🐛 Proposed fix for the substring issue
- existing = state.churn_risk_users or ""
- if segment_id not in existing:
- state.churn_risk_users = (existing + "," + segment_id).strip(",")
+ existing_list = [s for s in (state.churn_risk_users or "").split(",") if s]
+ if segment_id not in existing_list:
+ state.churn_risk_users = ",".join(existing_list + [segment_id])
state.last_updated_by = "flag_churn_risk_tool"
await update_mission_state(state)The concurrent-write race would still require an atomic DB-level append (e.g., array column or SELECT ... FOR UPDATE) to fully resolve.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| state = await get_mission_state(tenant_id) | |
| existing = state.churn_risk_users or "" | |
| if segment_id not in existing: | |
| state.churn_risk_users = (existing + "," + segment_id).strip(",") | |
| state.last_updated_by = "flag_churn_risk_tool" | |
| await update_mission_state(state) | |
| state = await get_mission_state(tenant_id) | |
| existing_list = [s for s in (state.churn_risk_users or "").split(",") if s] | |
| if segment_id not in existing_list: | |
| state.churn_risk_users = ",".join(existing_list + [segment_id]) | |
| state.last_updated_by = "flag_churn_risk_tool" | |
| await update_mission_state(state) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/ai/src/agents/tools/flag_churn_risk.py` around lines 36 - 41, The
`flag_churn_risk` flow is using raw substring checks on `state.churn_risk_users`
and a non-atomic read-modify-write update. Update the `flag_churn_risk` tool to
treat `churn_risk_users` as a parsed list of segment IDs (not a comma-joined
string) so membership is checked by exact element match, and change the
`get_mission_state`/`update_mission_state` path to an atomic append or locked
update so concurrent calls cannot overwrite each other’s additions. Use the
existing `flag_churn_risk` function and `update_mission_state` call site as the
main places to adjust.
| client = SlackClient() | ||
| channel = os.getenv("SLACK_CHECKIN_CHANNEL", "#customer-checkins") | ||
| msg = f"*Check-in reminder:* Follow up with customer `{customer_id}` in {days_out} days." | ||
| resp = client.client.chat_postMessage(channel=channel, text=msg) | ||
| return { | ||
| "scheduled": True, | ||
| "customer_id": customer_id, | ||
| "days_out": days_out, | ||
| "tenant_id": tenant_id, | ||
| "channel": channel, | ||
| "ts": resp.get("ts", ""), | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the target file and nearby context.
python3 - <<'PY'
from pathlib import Path
p = Path('apps/ai/src/agents/tools/schedule_customer_checkin.py')
print("FILE:", p)
print("LINES:", sum(1 for _ in p.open()))
for i, line in enumerate(p.open(), 1):
if 1 <= i <= 220:
print(f"{i:4}: {line.rstrip()}")
PY
# Find Slack client usage and async tool patterns.
rg -n "class SlackClient|AsyncWebClient|chat_postMessage|async def execute|to_thread|SlackClient\(" apps/ai/src -SRepository: Aparnap2/Track_Guard
Length of output: 5633
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the tool package to understand registration and whether async tools are expected.
ast-grep outline apps/ai/src/agents/tools --view expandedRepository: Aparnap2/Track_Guard
Length of output: 1565
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Read any relevant Slack client wrapper if found by search.
for f in $(rg -l "class SlackClient|AsyncWebClient" apps/ai/src -S); do
echo "==== $f ===="
wc -l "$f"
python3 - <<'PY' "$f"
import sys
from pathlib import Path
p = Path(sys.argv[1])
for i, line in enumerate(p.open(), 1):
if i <= 260:
print(f"{i:4}: {line.rstrip()}")
PY
doneRepository: Aparnap2/Track_Guard
Length of output: 11174
Avoid the blocking Slack call in async def execute.
SlackClient.client is a synchronous WebClient, so chat_postMessage() blocks the event loop while the request is in flight. Wrap it with asyncio.to_thread(...) here or switch this path to AsyncWebClient.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/ai/src/agents/tools/schedule_customer_checkin.py` around lines 46 - 57,
The `execute` path in `schedule_customer_checkin` is blocking the event loop
because `SlackClient.client` is a synchronous `WebClient` and
`chat_postMessage()` is called directly. Update the `execute` method to use a
non-blocking approach by either wrapping the `chat_postMessage` call with
`asyncio.to_thread(...)` or switching this flow to `AsyncWebClient`, and keep
the rest of the response assembly (including `resp.get("ts", "")`) unchanged.
| try: | ||
| # chat_completion is synchronous | ||
| brief = chat_completion( | ||
| messages=[ | ||
| {"role": "system", "content": "You are a concise business briefing assistant. Output exactly 2 sentences. No preamble."}, | ||
| {"role": "user", "content": prompt}, | ||
| ], | ||
| max_tokens=80, | ||
| temperature=0.3, | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm chat_completion is indeed a blocking, synchronous call and check for any timeout config.
rg -n 'def chat_completion' -A 20 apps/ai/src/config/llm.pyRepository: Aparnap2/Track_Guard
Length of output: 3460
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- apps/ai/src/config/llm.py: chat_completion and kwarg building ---'
sed -n '1,220p' apps/ai/src/config/llm.py
echo
echo '--- apps/ai/src/session/brief_generator.py ---'
sed -n '1,220p' apps/ai/src/session/brief_generator.py
echo
echo '--- caller context in mission_state.py ---'
sed -n '250,340p' apps/ai/src/session/mission_state.pyRepository: Aparnap2/Track_Guard
Length of output: 11449
Move the blocking LLM call off the event loop.
generate_prepared_brief() is async, but it calls the synchronous chat_completion() directly. That can block the shared event loop for up to the client timeout (60s), stalling other coroutines while the brief is generated. Use await asyncio.to_thread(...) here, or switch this path to an async LLM client.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/ai/src/session/brief_generator.py` around lines 29 - 38, The async flow
in generate_prepared_brief currently calls the synchronous chat_completion()
directly, which can block the shared event loop. Update this call site to run
chat_completion() off the loop using await asyncio.to_thread(...) or replace it
with an async LLM client, keeping the rest of the brief generation logic
unchanged.
| guardrail_blocking, investor_facing_alert, | ||
| prepared_brief, pending_decisions, last_updated_by, | ||
| last_update_reason, last_changed_fields, active_agent_roles, | ||
| policy_state |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether prepared_brief/pending_decisions/last_updated_by/policy_state
# columns are created anywhere in the migration history.
fd -e sql . apps/core/internal/db/migrations apps/ai/infrastructure/migrations 2>/dev/null -x cat {}Repository: Aparnap2/Track_Guard
Length of output: 15698
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== schema references ==\n'
rg -n "policy_state|prepared_brief|pending_decisions|last_updated_by|last_update_reason|last_changed_fields|active_agent_roles" .
printf '\n== mission_state.py around the relevant write/read paths ==\n'
sed -n '1,340p' apps/ai/src/session/mission_state.pyRepository: Aparnap2/Track_Guard
Length of output: 21081
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== policy_state references ==\n'
rg -n "policy_state" apps . --glob '!**/node_modules/**' --glob '!**/dist/**' --glob '!**/build/**'
printf '\n== migration files mentioning mission_states ==\n'
fd -a -e sql mission_state apps . 2>/dev/null | sort | xargs -r -I{} sh -c 'echo "--- {}"; rg -n "mission_states|policy_state|prepared_brief|pending_decisions|last_updated_by" "{}"'Repository: Aparnap2/Track_Guard
Length of output: 3796
Add a migration for policy_state. prepared_brief, pending_decisions, and last_updated_by already exist in 004_resolve_mission_state_drift.sql; policy_state is the only field here without a matching schema change, and both get_mission_state and update_mission_state depend on it.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/ai/src/session/mission_state.py` around lines 127 - 130, Add a schema
migration for policy_state in the mission state model, since it is referenced by
get_mission_state and update_mission_state but does not yet have a matching
schema change. Update the migration set that already covers the other
mission_state fields so policy_state is included in the database schema and any
relevant drift-fix SQL, using the mission_state data model as the locator for
the missing field. Ensure the migration is applied consistently with the
existing mission state schema changes.
| if generate_brief: | ||
| state.last_update_reason = update_reason | ||
| state.last_changed_fields = changed_fields |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
generate_brief flag incorrectly gates explainability attribution.
Setting state.last_update_reason/state.last_changed_fields from the function args is conditioned on generate_brief, which is unrelated:
- Callers passing
generate_brief=Falsetogether withupdate_reason/changed_fields(a natural pattern to record why while skipping LLM brief generation) have those values silently dropped. - Callers using the default
generate_brief=Truewithout passingupdate_reason/changed_fieldswill overwrite any value the caller had already set directly onstate(e.g.state.last_update_reason = "..."before calling) back toNone.
🐛 Proposed fix
- if generate_brief:
- state.last_update_reason = update_reason
- state.last_changed_fields = changed_fields
+ if update_reason is not None:
+ state.last_update_reason = update_reason
+ if changed_fields is not None:
+ state.last_changed_fields = changed_fields📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if generate_brief: | |
| state.last_update_reason = update_reason | |
| state.last_changed_fields = changed_fields | |
| if update_reason is not None: | |
| state.last_update_reason = update_reason | |
| if changed_fields is not None: | |
| state.last_changed_fields = changed_fields |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/ai/src/session/mission_state.py` around lines 196 - 198, The attribution
fields in mission state are being updated under the unrelated generate_brief
flag, which causes caller-provided update_reason/changed_fields to be dropped or
existing state values to be overwritten with None. In the mission_state update
logic, move the last_update_reason and last_changed_fields assignment out of the
generate_brief branch and only set them when the corresponding function
arguments are actually provided, preserving any values already on state; use the
surrounding update routine and the
state.last_update_reason/state.last_changed_fields fields to locate the fix.
| state.prepared_brief, | ||
| state.pending_decisions, |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the target area and related serialization patterns.
FILE="apps/ai/src/session/mission_state.py"
echo "== Outline =="
ast-grep outline "$FILE" --view expanded || true
echo
echo "== Target lines =="
nl -ba "$FILE" | sed -n '240,320p'
echo
echo "== Search for pending_decisions / JSONB / codec usage =="
rg -n "pending_decisions|last_changed_fields|active_agent_roles|policy_state|set_type_codec|json\.dumps|jsonb" apps/ai/src/session apps/ai/src -g '!**/__pycache__/**' || trueRepository: Aparnap2/Track_Guard
Length of output: 2222
🏁 Script executed:
#!/bin/bash
set -euo pipefail
FILE="apps/ai/src/session/mission_state.py"
echo "== update_mission_state section =="
sed -n '182,320p' "$FILE" | cat -n
echo
echo "== serialization-related matches in file =="
rg -n "json\.dumps|pending_decisions|last_changed_fields|active_agent_roles|policy_state|set_type_codec" "$FILE" || true
echo
echo "== repository-wide set_type_codec matches =="
rg -n "set_type_codec" apps/ai/src -g '!**/__pycache__/**' || trueRepository: Aparnap2/Track_Guard
Length of output: 8375
🌐 Web query:
asyncpg jsonb parameter list dict auto encode default codec documentation
💡 Result:
By default, asyncpg treats PostgreSQL json and jsonb types as strings, meaning it does not automatically encode Python dictionaries or decode them into dictionaries [1][2]. To achieve automatic conversion between Python objects (like dicts and lists) and PostgreSQL json/jsonb types, you must explicitly register a custom type codec using the Connection.set_type_codec method [3][1]. To register a codec for jsonb, use the following approach [1]: import json import asyncpg async def setup_connection(conn): # Register the codec for 'jsonb' await conn.set_type_codec( 'jsonb', encoder=json.dumps, decoder=json.loads, schema='pg_catalog') # Example usage with a connection pool pool = await asyncpg.create_pool( dsn="postgres://user:pass@host/db", init=setup_connection) When using this codec, asyncpg will automatically handle encoding your Python dictionaries when passing them as parameters and decoding jsonb columns back into Python objects when fetching results [4][1]. Note that the schema should be set to 'pg_catalog' for built-in types [3][1]. If you are using a connection pool, it is recommended to use the init argument in create_pool to ensure the codec is registered for all connections [2].
Citations:
- 1: https://github.com/MagicStack/asyncpg/blob/master/docs/usage.rst
- 2: json_agg and json_build_object raw queries return string type rather than json MagicStack/asyncpg#432
- 3: https://magicstack.github.io/asyncpg/devel/usage.html
- 4: JSONB positional parameter MagicStack/asyncpg#912
Serialize pending_decisions before binding it to jsonb.
asyncpg doesn’t auto-encode Python lists/dicts for jsonb parameters unless a custom codec is registered, so this raw list[dict] needs the same json.dumps(...) treatment as the other JSONB fields. Use json.dumps(state.pending_decisions) if state.pending_decisions else '[]'.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/ai/src/session/mission_state.py` around lines 277 - 278,
`pending_decisions` is being passed to an asyncpg jsonb parameter as a raw
Python list in the mission state save path, unlike the other JSONB fields.
Update the serialization in the mission-state persistence code that uses
`state.prepared_brief` and `state.pending_decisions` so `pending_decisions` is
converted with `json.dumps(state.pending_decisions) if state.pending_decisions
else '[]'` before binding, matching the handling used for the other jsonb
values.
| // Unsubscribe removes a subscription by ID | ||
| func (h *SSEHub) Unsubscribe(tenantID, subID string) { | ||
| h.mu.Lock() | ||
| defer h.mu.Unlock() | ||
| if subs, ok := h.subscribers[tenantID]; ok { | ||
| if ch, ok := subs[subID]; ok { | ||
| close(ch) | ||
| if sub, ok := subs[subID]; ok { | ||
| close(sub.Channel) | ||
| delete(subs, subID) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the target file before reading it
ast-grep outline apps/core/internal/web/sse_hub.go --view expanded || true
echo '--- FILE: apps/core/internal/web/sse_hub.go (first 220 lines) ---'
sed -n '1,220p' apps/core/internal/web/sse_hub.go
echo '--- SEARCH: tenant_id usage in apps/core/internal/web ---'
rg -n 'tenant_id|NewSSEHub|Unsubscribe\(|Subscribe\(' apps/core/internal/web -g '*.go'Repository: Aparnap2/Track_Guard
Length of output: 5554
Drop empty tenant buckets in Unsubscribe
delete(subs, subID) leaves an empty tenantID entry behind. Since tenant_id is caller-controlled and the hub is long-lived, repeated subscribe/unsubscribe cycles can accumulate unbounded empty buckets. Remove the tenant key when len(subs) == 0.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/core/internal/web/sse_hub.go` around lines 55 - 65, The Unsubscribe
method on SSEHub removes a subID but leaves an empty tenant bucket behind in the
subscribers map. Update Unsubscribe to delete the tenantID entry from
h.subscribers when the inner subs map becomes empty after delete(subs, subID),
using the existing SSEHub and Unsubscribe symbols to locate the cleanup logic.
Keep the channel close and subscription removal behavior the same, but add the
empty-map check before returning so tenant buckets do not accumulate.
1 participant
Summary
Lightweight control plane and risk-governed operating layer on top of Sarthi's existing guardian model, MissionState core, and specialist agents.
Changes
New Modules
PolicyDecision,RiskScanResult,AgentRegistration,AuditEventPydantic v2 contractsaudit_logtable with indexesModified
allowed_models,external_facing,data_classificationto all 5 agentspolicy_statefield with full DB roundtrip (SELECT/INSERT/UPDATE)requires_approval: TrueTests
test_control_plane.py— schemas, registry, policy engine, HITL enforcement, MissionState policy_stateArchitecture
Preserves Sarthi's guardian architecture, deterministic watchlists, bounded LLM usage, and narrow-workflow-first rollout pattern. External-facing outputs are always policy-evaluated, risk-scanned, audit-logged, and HITL-routed before release.
Summary by CodeRabbit
New Features
Bug Fixes