feat: runtime profiles + local CI infrastructure#29
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).
Add three Docker Compose profiles matching machine constraints: - dev: PostgreSQL + Redis only (~2GB, daily coding) - llmops: adds Jaeger + MinIO + Langfuse (~4-5GB, tracing/eval) - showcase: full stack Temporal + Redpanda + Qdrant + Langfuse (~9GB, demo only) Container names and volumes are profile-scoped to prevent collisions. All profiles share iterateswarm-net for compatible service discovery.
Add local-first CI with pre-push validation: - make ci-fast: vet + fmt + ruff + unit tests (pre-commit gate) - make ci-local: ci-fast + actionlint (pre-push gate) - Pre-push hook: blocks direct pushes to main, runs actionlint - GitHub Actions CI: typecheck -> lint -> unit-tests + workflow-lint - Integration test job exists as disabled placeholder Also fix pre-existing bugs discovered by local gates: - Fix orphaned except in hubspot.py (SyntaxError) - Fix missing try: block in quickbooks.py (SyntaxError) - Add missing lineage field in test_agentic_comprehensive.py - Fix broad "llm" substring match in test_qa_agent.py - Ignore env-var-dependent and Docker-dependent tests in ci-fast Config drift is now caught locally before push.
📝 WalkthroughWalkthroughThis PR delivers a v4 architecture evolution: typed SSEHub fan-out, a new control-plane (registry/policy/audit) with a global ToolRegistry of HITL-tiered tools guarded by prompt/output risk scanning, Slack SocketMode with an ACE feedback loop, MissionState explainability fields and a prepared-brief generator, plus a rewritten CI pipeline, git hooks, docker-compose profiles, and extensive documentation updates. ChangesSarthi V4 Backend Features
CI, Local Dev Infra, and Documentation
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Agent
participant ToolRegistry
participant RiskScanner
participant PolicyEngine
participant AuditLogger
Agent->>ToolRegistry: get_tools_for_pattern(pattern_id)
ToolRegistry-->>Agent: ToolDef
Agent->>PolicyEngine: evaluate(agent, requested_tool)
PolicyEngine-->>Agent: PolicyDecision
Agent->>RiskScanner: scan_prompt(prompt)
RiskScanner-->>Agent: RiskScanResult
Agent->>Agent: execute tool (draft/flag/pause/schedule)
Agent->>RiskScanner: scan_output(result)
RiskScanner-->>Agent: RiskScanResult
Agent->>AuditLogger: log_event(AuditEvent)
sequenceDiagram
participant Slack
participant SlackClient
participant SlackButtons
participant Curator
participant MissionState
Slack->>SlackClient: interactive button payload (SocketMode)
SlackClient->>SlackButtons: route_slack_button()
SlackButtons->>SlackButtons: score_from_button()
SlackButtons->>Curator: update_strategy_confidence()
Curator-->>SlackButtons: (ConfidenceUpdateResult, StrategyDelta)
SlackButtons->>Curator: curator.update(strategy, score_delta)
Curator->>MissionState: update_mission_state(state)
MissionState-->>Curator: persisted with prepared_brief
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 several major architectural updates to the Sarthi platform, including a centralized control plane with deterministic policy checks and risk scanning, a tool calling surface mapped to HITL tiers, Slack SocketMode integration with an ACE feedback loop, and an event-filtered SSEHub for typed streaming. However, several critical issues were identified: in handler.go, multiple SSE endpoints incorrectly defer Unsubscribe in the handler rather than the background stream writer, causing immediate disconnects, and queries still reference the deprecated singular mission_state table instead of mission_states. Additionally, an indentation error in quickbooks.py will cause a runtime NameError, _write_audit_log in curator.py fails to pass connection arguments to asyncpg.connect() while also breaking the file fallback mechanism, and the risk scanners in output_risk.py and prompt_risk.py incorrectly use re.findall on patterns with capturing groups, resulting in incomplete matched text.
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.
| c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { | ||
| defer func() { recover() }() |
There was a problem hiding this comment.
The Unsubscribe call must be deferred inside the background stream writer function so that it executes only when the stream writer exits.
| c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { | |
| defer func() { recover() }() | |
| c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { | |
| defer h.sseHub.Unsubscribe(tenantID, sub.ID) | |
| defer func() { recover() }() |
| func (h *Handler) APICommandChatEvents(c *fiber.Ctx) error { | ||
| tenantID := c.Query("tenant_id", "default") | ||
| subID, ch := h.sseHub.Subscribe(tenantID) | ||
| defer h.sseHub.Unsubscribe(tenantID, subID) | ||
| sub := h.sseHub.Subscribe(tenantID, "chat") | ||
| defer h.sseHub.Unsubscribe(tenantID, sub.ID) |
There was a problem hiding this comment.
Deferring Unsubscribe inside the handler function causes it to execute immediately when the handler returns nil (which happens right after registering the stream writer). This closes the subscription channel before the background stream writer can read from it, causing the SSE stream to disconnect instantly.
| func (h *Handler) APICommandChatEvents(c *fiber.Ctx) error { | |
| tenantID := c.Query("tenant_id", "default") | |
| subID, ch := h.sseHub.Subscribe(tenantID) | |
| defer h.sseHub.Unsubscribe(tenantID, subID) | |
| sub := h.sseHub.Subscribe(tenantID, "chat") | |
| defer h.sseHub.Unsubscribe(tenantID, sub.ID) | |
| func (h *Handler) APICommandChatEvents(c *fiber.Ctx) error { | |
| tenantID := c.Query("tenant_id", "default") | |
| sub := h.sseHub.Subscribe(tenantID, "chat") |
| func (h *Handler) APICommandMissionEvents(c *fiber.Ctx) error { | ||
| tenantID := c.Query("tenant_id", "default") | ||
| sub := h.sseHub.Subscribe(tenantID, "mission-update") | ||
| defer h.sseHub.Unsubscribe(tenantID, sub.ID) |
There was a problem hiding this comment.
Deferring Unsubscribe inside the handler function causes it to execute immediately when the handler returns nil, closing the subscription channel before the background stream writer can read from it.
| func (h *Handler) APICommandMissionEvents(c *fiber.Ctx) error { | |
| tenantID := c.Query("tenant_id", "default") | |
| sub := h.sseHub.Subscribe(tenantID, "mission-update") | |
| defer h.sseHub.Unsubscribe(tenantID, sub.ID) | |
| func (h *Handler) APICommandMissionEvents(c *fiber.Ctx) error { | |
| tenantID := c.Query("tenant_id", "default") | |
| sub := h.sseHub.Subscribe(tenantID, "mission-update") |
| c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { | ||
| defer func() { recover() }() |
There was a problem hiding this comment.
The Unsubscribe call must be deferred inside the background stream writer function so that it executes only when the stream writer exits.
| c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { | |
| defer func() { recover() }() | |
| c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { | |
| defer h.sseHub.Unsubscribe(tenantID, sub.ID) | |
| defer func() { recover() }() |
| func (h *Handler) APICommandHITLEvents(c *fiber.Ctx) error { | ||
| tenantID := c.Query("tenant_id", "default") | ||
| sub := h.sseHub.Subscribe(tenantID, "hitl-item") | ||
| defer h.sseHub.Unsubscribe(tenantID, sub.ID) |
There was a problem hiding this comment.
Deferring Unsubscribe inside the handler function causes it to execute immediately when the handler returns nil, closing the subscription channel before the background stream writer can read from it.
| func (h *Handler) APICommandHITLEvents(c *fiber.Ctx) error { | |
| tenantID := c.Query("tenant_id", "default") | |
| sub := h.sseHub.Subscribe(tenantID, "hitl-item") | |
| defer h.sseHub.Unsubscribe(tenantID, sub.ID) | |
| func (h *Handler) APICommandHITLEvents(c *fiber.Ctx) error { | |
| tenantID := c.Query("tenant_id", "default") | |
| sub := h.sseHub.Subscribe(tenantID, "hitl-item") |
| COALESCE(runway_days, 0), | ||
| last_update_reason, | ||
| last_changed_fields::text, | ||
| active_agent_roles::text | ||
| FROM mission_state | ||
| ORDER BY updated_at DESC |
There was a problem hiding this comment.
The query references FROM mission_state (singular), but the table was renamed to mission_states (plural) in migration 004. This mismatch will cause a database error at runtime.
| COALESCE(runway_days, 0), | |
| last_update_reason, | |
| last_changed_fields::text, | |
| active_agent_roles::text | |
| FROM mission_state | |
| ORDER BY updated_at DESC | |
| COALESCE(runway_days, 0), | |
| last_update_reason, | |
| last_changed_fields::text, | |
| active_agent_roles::text | |
| FROM mission_states | |
| ORDER BY updated_at DESC |
| pending_decisions::text, | ||
| active_agent_roles::text | ||
| FROM mission_state | ||
| ORDER BY updated_at DESC |
There was a problem hiding this comment.
The query references FROM mission_state (singular), but the table was renamed to mission_states (plural) in migration 004. This mismatch will cause a database error at runtime.
| pending_decisions::text, | |
| active_agent_roles::text | |
| FROM mission_state | |
| ORDER BY updated_at DESC | |
| pending_decisions::text, | |
| active_agent_roles::text | |
| FROM mission_states | |
| ORDER BY updated_at DESC |
| def _write_audit_log(delta: StrategyDelta) -> None: | ||
| """Append StrategyDelta to audit log. Tries PostgreSQL, falls back to file.""" | ||
| try: | ||
| import asyncpg # noqa: F401 — available if postgres is configured | ||
| import asyncio | ||
|
|
||
| async def _pg_write() -> None: | ||
| conn = await asyncpg.connect() | ||
| await conn.execute(""" | ||
| CREATE TABLE IF NOT EXISTS strategy_deltas ( | ||
| id SERIAL PRIMARY KEY, | ||
| tenant_id TEXT NOT NULL, | ||
| domain TEXT NOT NULL, | ||
| strategy_key TEXT NOT NULL, | ||
| prior_confidence DOUBLE PRECISION NOT NULL, | ||
| new_confidence DOUBLE PRECISION NOT NULL, | ||
| evidence_count_delta INTEGER NOT NULL, | ||
| feedback_type TEXT NOT NULL, | ||
| linked_alert_id TEXT NOT NULL, | ||
| created_at TIMESTAMPTZ NOT NULL DEFAULT now() | ||
| ) | ||
| """) | ||
| await conn.execute( | ||
| """ | ||
| INSERT INTO strategy_deltas | ||
| (tenant_id, domain, strategy_key, prior_confidence, | ||
| new_confidence, evidence_count_delta, feedback_type, | ||
| linked_alert_id) | ||
| VALUES ($1, $2, $3, $4, $5, $6, $7, $8) | ||
| """, | ||
| delta.tenant_id, | ||
| delta.domain, | ||
| delta.strategy_key, | ||
| delta.prior_confidence, | ||
| delta.new_confidence, | ||
| delta.evidence_count_delta, | ||
| delta.feedback_type, | ||
| delta.linked_alert_id, | ||
| ) | ||
| await conn.close() | ||
|
|
||
| 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.
The _write_audit_log function fails to connect to PostgreSQL because asyncpg.connect() is called with no arguments. Additionally, the fallback mechanism is broken because any non-ImportError exception (such as connection or query execution errors) is re-raised and not caught by the outer except ImportError block, preventing the fallback file write from executing.
def _write_audit_log(delta: StrategyDelta) -> None:
"""Append StrategyDelta to audit log. Tries PostgreSQL, falls back to file."""
try:
import asyncpg
import asyncio
from src.config.database import get_database_url
async def _pg_write() -> None:
conn = await asyncpg.connect(get_database_url("iterateswarm"))
await conn.execute("""
CREATE TABLE IF NOT EXISTS strategy_deltas (
id SERIAL PRIMARY KEY,
tenant_id TEXT NOT NULL,
domain TEXT NOT NULL,
strategy_key TEXT NOT NULL,
prior_confidence DOUBLE PRECISION NOT NULL,
new_confidence DOUBLE PRECISION NOT NULL,
evidence_count_delta INTEGER NOT NULL,
feedback_type TEXT NOT NULL,
linked_alert_id TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
)
""")
await conn.execute(
"""
INSERT INTO strategy_deltas
(tenant_id, domain, strategy_key, prior_confidence,
new_confidence, evidence_count_delta, feedback_type,
linked_alert_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
""",
delta.tenant_id,
delta.domain,
delta.strategy_key,
delta.prior_confidence,
delta.new_confidence,
delta.evidence_count_delta,
delta.feedback_type,
delta.linked_alert_id,
)
await conn.close()
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:
_file_audit_write(delta)| for rule_id, pattern, severity in _ALL_OUTPUT_PATTERNS: | ||
| matches = re.findall(pattern, text) | ||
| if matches: | ||
| matched_text = str(matches[0]) if matches else "" | ||
| if len(matched_text) > 80: | ||
| matched_text = matched_text[:77] + "..." | ||
| flags.append(RiskFlag( | ||
| rule_id=rule_id, | ||
| description=_get_output_rule_description(rule_id), | ||
| severity=severity, | ||
| matched_text=matched_text, | ||
| )) |
There was a problem hiding this comment.
Using re.findall on patterns with capturing groups returns only the captured subgroups rather than the full matched text. This causes matched_text to be incomplete or formatted as a tuple string. Using re.search is more correct and efficient.
| for rule_id, pattern, severity in _ALL_OUTPUT_PATTERNS: | |
| matches = re.findall(pattern, text) | |
| if matches: | |
| matched_text = str(matches[0]) if matches else "" | |
| if len(matched_text) > 80: | |
| matched_text = matched_text[:77] + "..." | |
| flags.append(RiskFlag( | |
| rule_id=rule_id, | |
| description=_get_output_rule_description(rule_id), | |
| severity=severity, | |
| matched_text=matched_text, | |
| )) | |
| for rule_id, pattern, severity in _ALL_OUTPUT_PATTERNS: | |
| match = re.search(pattern, text) | |
| if match: | |
| matched_text = match.group(0) | |
| if len(matched_text) > 80: | |
| matched_text = matched_text[:77] + "..." | |
| flags.append(RiskFlag( | |
| rule_id=rule_id, | |
| description=_get_output_rule_description(rule_id), | |
| severity=severity, | |
| matched_text=matched_text, | |
| )) |
| for rule_id, pattern, severity in _ALL_PATTERNS: | ||
| matches = re.findall(pattern, text) | ||
| if matches: | ||
| matched_text = str(matches[0]) if matches else "" | ||
| # Truncate matched text for display | ||
| if len(matched_text) > 80: | ||
| matched_text = matched_text[:77] + "..." | ||
| flags.append(RiskFlag( | ||
| rule_id=rule_id, | ||
| description=_get_rule_description(rule_id), | ||
| severity=severity, | ||
| matched_text=matched_text, | ||
| )) |
There was a problem hiding this comment.
Using re.findall on patterns with capturing groups returns only the captured subgroups rather than the full matched text. This causes matched_text to be incomplete or formatted as a tuple string. Using re.search is more correct and efficient.
| for rule_id, pattern, severity in _ALL_PATTERNS: | |
| matches = re.findall(pattern, text) | |
| if matches: | |
| matched_text = str(matches[0]) if matches else "" | |
| # Truncate matched text for display | |
| if len(matched_text) > 80: | |
| matched_text = matched_text[:77] + "..." | |
| flags.append(RiskFlag( | |
| rule_id=rule_id, | |
| description=_get_rule_description(rule_id), | |
| severity=severity, | |
| matched_text=matched_text, | |
| )) | |
| for rule_id, pattern, severity in _ALL_PATTERNS: | |
| match = re.search(pattern, text) | |
| if match: | |
| matched_text = match.group(0) | |
| # Truncate matched text for display | |
| if len(matched_text) > 80: | |
| matched_text = matched_text[:77] + "..." | |
| flags.append(RiskFlag( | |
| rule_id=rule_id, | |
| description=_get_rule_description(rule_id), | |
| severity=severity, | |
| matched_text=matched_text, | |
| )) |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Note
Due to the large number of review comments, Critical 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 (3)
apps/ai/src/integrations/quickbooks.py (1)
86-172: 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick winFix the invoice fetch path —
query_responseis scoped inside_fetch_invoices()but never returned or called, soinvoices = query_response.get(...)raisesNameErroron the first real request. The broadexceptthen swallows it and returns_MOCK_DATA, so production always gets mock QuickBooks metrics. Replace it withinvoices = _fetch_invoices()or return the invoice list from the helper, and narrow the exception handling so real API failures aren’t silently masked.🤖 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/quickbooks.py` around lines 86 - 172, The invoice fetch flow in quickbooks.py is broken because _fetch_invoices() defines query_response locally but never returns invoices, while the outer scope still reads query_response directly, causing a NameError and triggering the fallback to mock data. Update _fetch_invoices() to return the parsed Invoice list (or assign invoices from its return value) and make the main QuickBooks fetch logic use that result before the metrics aggregation loop. Also tighten the broad exception handling around the QuickBooks request path so real HTTP or parsing failures are surfaced instead of silently returning _MOCK_DATA.apps/ai/src/session/mission_state.py (1)
121-130: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDecode the JSONB fields before assigning them to
MissionState.asyncpgreturns these columns as raw JSON text unless a JSON codec is registered, sopending_decisions,last_changed_fields, andactive_agent_rolesneedjson.loads(...)here (or a connection-level codec). Otherwise theselist/dictfields will hold strings instead of Python objects.🤖 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 121 - 130, The MissionState query is assigning JSONB-backed columns as raw strings instead of Python objects. Update the fetch/mapping logic in MissionState to decode pending_decisions, last_changed_fields, and active_agent_roles with json.loads(...) before constructing the MissionState instance, or register a JSON codec on the asyncpg connection so these fields are returned decoded.apps/core/hitl_queue_test.go (1)
132-151: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winOverly broad substring check for
"rate"risks false-positive passes.
strings.Contains(bodyLower, "rate")will match unrelated substrings like "generate", "integrate", "separate", "calibrate", etc. This is the same class of brittle-substring issue this PR fixes elsewhere intest_qa_agent.py. If the llmops partial ever contains any of those unrelated words, the test passes even if actual metrics content is missing/broken.🔧 Suggested fix: use more specific phrase checks
- hasMetric := strings.Contains(bodyLower, "score") || strings.Contains(bodyLower, "rate") || strings.Contains(bodyLower, "quality") + hasMetric := strings.Contains(bodyLower, "score") || strings.Contains(bodyLower, "success rate") || strings.Contains(bodyLower, "quality")🤖 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/hitl_queue_test.go` around lines 132 - 151, The llmops HTMX partial test uses an overly broad substring check in TestLLMOpsDashboardReturnsHTMXPartial, where strings.Contains(bodyLower, "rate") can pass on unrelated words. Tighten the assertion by replacing that generic match with a more specific metrics phrase or token check, using the existing score/quality checks as guidance, so the test only passes when the LLM ops partial actually contains the intended metrics content.
🟠 Major comments (18)
apps/core/internal/web/handler.go-1582-1585 (1)
1582-1585: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winMove
Unsubscribeinto the stream writer.defer h.sseHub.Unsubscribe(tenantID, sub.ID)runs when the handler returns, so the SSE channel closes beforeSetBodyStreamWriterstarts. That leaves clients stuck onconnected/reconnect churn. Apply the same change inAPICommandChatEvents,APICommandMissionEvents,APICommandHITLEvents, andAPICommandSessionEvents.🤖 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 1582 - 1585, Move the h.sseHub.Unsubscribe call out of the handler defer and into the SetBodyStreamWriter lifecycle so the subscription stays active until the SSE stream ends. Update APICommandChatEvents, APICommandMissionEvents, APICommandHITLEvents, and APICommandSessionEvents to unsubscribe inside the stream writer cleanup logic, using the existing tenantID and sub.ID values..github/workflows/ci.yml-18-18 (1)
18-18: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
actions/checkoutsteps don't setpersist-credentials: false.Static analysis (zizmor) flags credential persistence risk across all 5 checkout steps. Since no job needs to push using the checked-out token, disable credential persistence.
🔒 Proposed fix (repeat for each checkout step)
- - uses: actions/checkout@v4 + - uses: actions/checkout@v4 + with: + persist-credentials: falseAlso applies to: 38-38, 62-62, 82-82, 93-93
🤖 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 @.github/workflows/ci.yml at line 18, The workflow’s multiple actions/checkout uses are persisting credentials by default, which triggers the security finding. Update each checkout step in the CI workflow to explicitly disable token persistence by setting persist-credentials to false, and apply the same change to every checkout occurrence referenced by the review so none of the jobs retain the checked-out token unnecessarily.Source: Linters/SAST tools
.github/workflows/ci.yml-14-97 (1)
14-97: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winNo workflow/job-level
permissions:block — default (broad) permissions used.zizmor flags excessive-permissions on all four jobs. Add a least-privilege
permissions:block (e.g.contents: read, pluspull-requests: write/checks: writeonly whereworkflow-lint's reviewdog action needs to post annotations).🤖 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 @.github/workflows/ci.yml around lines 14 - 97, The workflow jobs currently rely on the default broad GitHub token permissions, which triggers excessive-permissions warnings. Add least-privilege permissions blocks to the jobs in the CI workflow: keep checkout and test jobs limited to contents: read, and grant workflow-lint only the extra permissions needed by reviewdog/action-actionlint to publish annotations (such as checks: write and/or pull-requests: write). Use the existing job names typecheck, lint, unit-tests, workflow-lint, and integration-tests to place the permissions appropriately.Source: Linters/SAST tools
Makefile-414-415 (1)
414-415: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winSame Ruff masking issue as
ci.yml.
ruff check src/ 2>/dev/null && echo "ruff OK" || echo "ruff not configured, skipping"treats real lint failures the same as "ruff not installed" —ci-fastcan never fail on Ruff violations.🤖 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 `@Makefile` around lines 414 - 415, The Ruff step in the Makefile currently hides real lint failures by treating any nonzero exit from the `uv run ruff check` command the same as Ruff being unavailable. Update the `ci-fast` Ruff check logic in this block so it only skips when Ruff is actually missing/unconfigured, and still fails the target when `ruff check src/` reports violations; keep the behavior aligned with the `ci.yml` Ruff step and use the existing `uv run ruff check`/`ruff OK`/`ruff not configured, skipping` flow as the place to fix it.Makefile-410-411 (1)
410-411: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winSame
go vetmasking issue asci.yml.
go vet ./... 2>&1 | grep -v 'gen/go' || truecan never fail:grepexits0exactly when it does find vet output, and|| trueforces success regardless.🤖 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 `@Makefile` around lines 410 - 411, The current go vet step is masking failures because the pipeline always succeeds; update the Makefile vet target so it can fail when vet reports real issues while still ignoring gen/go noise. Adjust the logic around the go vet invocation in the Makefile target (the block that echoes "Go vet..." and runs cd apps/core && go vet ./... 2>&1 | grep -v 'gen/go' || true) so the exit status reflects vet errors instead of being forced to success, and keep the filtering behavior tied to the vet command rather than the final pipeline result..github/workflows/ci.yml-30-32 (1)
30-32: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
mypyfailures are unconditionally suppressed.
uv run mypy src/ --ignore-missing-imports || truemasks every mypy error, so the "Type Check" job cannot fail on real type errors.🔧 Proposed fix
- cd apps/ai && uv sync --group dev && uv run mypy src/ --ignore-missing-imports || true + cd apps/ai && uv sync --group dev && uv run mypy src/ --ignore-missing-imports🤖 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 @.github/workflows/ci.yml around lines 30 - 32, The Python mypy step is swallowing all type-check failures because the `Python mypy` workflow command in the CI job ends with `|| true`. Remove that suppression so `uv run mypy src/ --ignore-missing-imports` can fail the job on real errors, and keep the existing `cd apps/ai && uv sync --group dev` setup unchanged..github/workflows/ci.yml-50-55 (1)
50-55: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winRuff check/format failures are indistinguishable from "not configured".
Both
ruff checkandruff format --checkfall through toecho "... not configured, skipping"on any nonzero exit, whether ruff is missing or actually reports violations. This means the lint job can never fail due to real Ruff findings.🔧 Proposed fix
- - name: Ruff check - run: | - cd apps/ai && uv run ruff check src/ || echo "ruff not configured, skipping" - - name: Ruff format - run: | - cd apps/ai && uv run ruff format --check src/ || echo "ruff format not configured, skipping" + - name: Ruff check + run: | + cd apps/ai && command -v ruff >/dev/null 2>&1 && uv run ruff check src/ || echo "ruff not configured, skipping" + - name: Ruff format + run: | + cd apps/ai && command -v ruff >/dev/null 2>&1 && uv run ruff format --check src/ || echo "ruff format not configured, skipping"🤖 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 @.github/workflows/ci.yml around lines 50 - 55, The Ruff steps in the CI workflow are swallowing all nonzero exits, so real lint/format violations are treated the same as Ruff being unavailable. Update the Ruff check and Ruff format steps in the workflow to only skip when the Ruff command itself is missing or unavailable, and allow actual `ruff check` / `ruff format --check` failures to fail the job. Use the existing `Ruff check` and `Ruff format` steps as the place to split “not configured” handling from genuine lint errors..github/workflows/ci.yml-22-24 (1)
22-24: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
go vetfailures are always swallowed.
go vet ... | grep -v 'gen/go' || true— the pipeline's exit status comes fromgrep, which exits0when it found matching (i.e. real, non-generated) vet issues, and1only when there were none. The trailing|| truethen forces success unconditionally either way. This typecheck step can never fail on a realgo vetfinding.🔧 Proposed fix
- - name: Go vet - run: | - cd apps/core && go vet ./... 2>&1 | grep -v 'gen/go' || true + - name: Go vet + run: | + cd apps/core && go vet ./... 2>&1 | grep -v 'gen/go' | tee /tmp/vet.out + test ! -s /tmp/vet.out🤖 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 @.github/workflows/ci.yml around lines 22 - 24, The Go vet step is currently masking all real failures because the `go vet` pipeline in the CI workflow always ends with success. Update the `Go vet` command so it preserves the exit status from `go vet` while still filtering out generated `gen/go` output, and remove the unconditional success behavior. Keep the fix localized to the CI job definition for the `Go vet` step so that real vet issues cause the workflow to fail.apps/ai/src/agents/tools/draft_investor_update.py-62-71 (1)
62-71: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winNo audit event is logged when the prompt-risk scan blocks.
Unlike the output-scan-block path (lines 82-104) and the success path (lines 107-120), this early return logs nothing to the audit trail. Given the migration's stated purpose — "Control plane audit trail for all agent actions gated by policy, risk, and HITL routing" — a blocked prompt should also be recorded for compliance/traceability.
🤖 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 62 - 71, The prompt-risk block path in draft_investor_update currently returns early after logging only to the runtime logger, so it never creates an audit record. Add an audit event in the prompt_scan.recommended_action == "block" branch, using the same audit logging mechanism/pattern already used in the output-scan block and success paths, so blocked prompt scans are recorded alongside tenant_id, scan_result, and requires_approval. Keep the fix localized to draft_investor_update and reuse the existing audit trail helper or logger symbols from the surrounding flow.apps/ai/src/risk/output_risk.py-41-45 (1)
41-45: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winOC009 fires backwards: flags "missing disclaimer" when the disclaimer is present.
_MISSING_APPROVALis folded into_ALL_OUTPUT_PATTERNS(line 45). The generic scan loop (67-78) then raises an OC009 flag — described as"Output is missing draft/pending-review disclaimer"— whenever the disclaimer pattern (draft|for review|pending approval|...) actually matches the text. This is the inverse of the intended behavior; the correct "missing disclaimer" case is already handled separately and correctly at lines 80-87 viahas_draft_disclaimer. As written, every properly-disclaimed draft gets an erroneous LOW-severity risk flag with a misleading description.🐛 Proposed fix
-_ALL_OUTPUT_PATTERNS = _UNSUPPORTED_CLAIMS + _PROMISE_PATTERNS + _PRICING_PATTERNS + _INVESTOR_MISSTATEMENTS + _MISSING_APPROVAL +_ALL_OUTPUT_PATTERNS = _UNSUPPORTED_CLAIMS + _PROMISE_PATTERNS + _PRICING_PATTERNS + _INVESTOR_MISSTATEMENTSAlso applies to: 62-78
🤖 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/risk/output_risk.py` around lines 41 - 45, OC009 is being triggered in the wrong direction because the `_MISSING_APPROVAL` pattern is included in `_ALL_OUTPUT_PATTERNS`, causing the generic scan in `output_risk.py` to flag texts that actually contain a draft/pending-review disclaimer. Update the logic around `_MISSING_APPROVAL`, `_ALL_OUTPUT_PATTERNS`, and the scan in the main risk evaluation so OC009 is emitted only when the disclaimer is absent, while the existing `has_draft_disclaimer` check remains the source of truth for this case. Make sure the OC009 description matches the actual condition being detected.apps/ai/src/agents/tools/flag_churn_risk.py-36-41 (2)
36-41: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winSubstring membership check on comma-joined string is unreliable.
if segment_id not in existingperforms a raw substring test against the wholechurn_risk_usersstring, not a delimiter-aware membership check. E.g. flagging"seg1"would be treated as already-present if"seg12"exists (false negative — never added), and conversely"seg12"could get skipped if"seg1"is a substring match. Split on the delimiter and check exact membership instead. This mirrors the substring-match issue the PR already fixed intest_qa_agent.py.🐛 Proposed fix
- existing = state.churn_risk_users or "" - if segment_id not in existing: - state.churn_risk_users = (existing + "," + segment_id).strip(",") + existing_ids = [s for s in (state.churn_risk_users or "").split(",") if s] + if segment_id not in existing_ids: + state.churn_risk_users = ",".join(existing_ids + [segment_id])🤖 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 membership check in flag_churn_risk.py is using a raw substring test on the comma-joined churn_risk_users value, which can incorrectly treat partial matches as existing entries. Update the logic in the flag_churn_risk tool to parse existing into individual IDs (using the same comma delimiter) and check exact membership before appending, then keep the state update flow in get_mission_state/update_mission_state unchanged.
36-41: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftAvoid overwriting
churn_risk_userson concurrent auto runs
get_mission_state()+update_mission_state()still performs a full-row read/modify/write onMissionState, andupdate_mission_state()persistschurn_risk_users = EXCLUDED.churn_risk_users. Twoflag_churn_risk_customerexecutions for the same tenant can race and silently drop one segment ID. Merge this field atomically in the database or add optimistic locking/version checks before writing.🤖 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 current read/modify/write flow in flag_churn_risk_customer using get_mission_state() and update_mission_state() can overwrite churn_risk_users when two runs race. Change the MissionState update path so churn_risk_users is merged atomically in the database or guarded with optimistic locking/version checks, and make the persistence logic in update_mission_state() avoid blindly replacing the field with EXCLUDED.churn_risk_users.apps/ai/src/schemas/control_plane.py-49-49 (1)
49-49: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winTighten
data_classification/approval_statetyping to match documented closed sets.
PolicyDecision.data_classification,AgentRegistration.data_classification, andAuditEvent.approval_stateare typed as plainstr/str | None, but their docstrings enumerate fixed value sets, and every sibling field in this file (domain,escalation_tier,health_status,action,outcome,status,recommended_action) usesLiteral[...]. This inconsistency lets typos or unexpected values pass Pydantic validation silently. It's especially risky fordata_classificationsincePolicyEngine._RESTRICTED_CLASSIFICATIONS(policy.py) gates restricted-data blocking on an exact string match — an unvalidated typo (e.g."Restrcted") would silently bypass the block instead of raising a validation error.🔧 Proposed fix
+DataClassification = Literal["internal", "external_investor", "external_customer", "restricted", "pii", "confidential"] + class PolicyDecision(BaseModel): ... - data_classification: str + data_classification: DataClassification ... class AuditEvent(BaseModel): ... - approval_state: str | None = None + approval_state: Literal["auto", "review", "approve", "blocked"] | None = None ... class AgentRegistration(BaseModel): ... - data_classification: str = "internal" + data_classification: DataClassification = "internal"Also applies to: 76-76, 94-94
🤖 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` at line 49, Tighten the typing for the closed-set fields in the control plane schemas by changing PolicyDecision.data_classification, AgentRegistration.data_classification, and AuditEvent.approval_state from plain str/optional str to Literal-based enums that match their documented allowed values, consistent with the other schema fields in this module. Update the corresponding model definitions in control_plane.py so Pydantic rejects invalid values instead of accepting typos, and keep the field names unchanged so the existing PolicyDecision, AgentRegistration, and AuditEvent references remain valid.apps/ai/src/control_plane/policy.py-74-97 (1)
74-97: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
blocked_reasonconflates "review-only" and "actually blocked" states, and can silently drop reasons.
blocked_reasonis set for the degraded-health case even though the action isn't actually blocked (per the comment "forces review but not full block"). This has two knock-on effects:
- Step 6 zeroes
allowed_model_classeswheneverblocked_reason is not None, so a merely-degraded agent loses all model access — contradicting the stated intent that it should still be able to act under review.- Step 5 only records the
tool_not_allowedreasonif blocked_reason is None, so a degraded agent requesting a disallowed tool silently loses the tool-not-allowed context in the audit trail.🔧 Proposed fix
- # 4. Degraded health — forces review but not full block - if agent.health_status == "degraded": - requires_human_approval = True - blocked_reason = "agent_health_degraded" + # 4. Degraded health — forces review but not full block + review_reasons: list[str] = [] + if agent.health_status == "degraded": + requires_human_approval = True + review_reasons.append("agent_health_degraded") # 5. Tool permission check if requested_tool is not None: if requested_tool in agent.allowed_tools: approved_tools.append(requested_tool) else: - if blocked_reason is None: - blocked_reason = f"tool_not_allowed: {requested_tool}" + review_reasons.append(f"tool_not_allowed: {requested_tool}") requires_human_approval = True # 6. Determine allowed model classes allowed_models = agent.allowed_models if blocked_reason is not None: allowed_models = [] return PolicyDecision( data_classification=agent.data_classification, allowed_model_classes=allowed_models, requires_human_approval=requires_human_approval, - blocked_reason=blocked_reason, + blocked_reason=blocked_reason or ("; ".join(review_reasons) or None), approved_tools=approved_tools, )🤖 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/policy.py` around lines 74 - 97, Separate “requires review” from “blocked” in PolicyDecision so degraded health does not clear access. In policy.py’s decision flow, keep agent.health_status == "degraded" setting requires_human_approval, but do not use blocked_reason for that state; reserve blocked_reason for true denials only. Update the tool check and allowed_models logic in the same function so tool_not_allowed can still be recorded for audit even when review is already required, and only empty allowed_model_classes when an actual block occurs.apps/ai/src/control_plane/audit.py-34-59 (1)
34-59: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winAdd a timeout on the DB connect/execute path.
asyncpg.connect(DATABASE_URL)and the subsequentconn.execute(...)have no timeout. A slow or unreachable database would hang the calling task indefinitely, since audit logging appears to be invoked inline from tool-execution flows.🔧 Proposed fix
- if own_conn: - conn = await asyncpg.connect(DATABASE_URL) + if own_conn: + conn = await asyncio.wait_for(asyncpg.connect(DATABASE_URL), timeout=5.0) - await conn.execute( + await asyncio.wait_for(conn.execute( """ INSERT INTO audit_log (...) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NOW()) """, ... - ) + ), timeout=5.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/ai/src/control_plane/audit.py` around lines 34 - 59, Add a timeout around the audit database path in audit_event so tool execution cannot hang indefinitely. Update the asyncpg.connect(DATABASE_URL) call and the subsequent conn.execute(...) in audit.py to use a bounded timeout (either via asyncpg timeout parameters or an async timeout wrapper) and ensure the same limit applies whether audit_event creates its own connection or reuses one. Keep the change localized to audit_event and its own_conn branch so the audit insert fails fast on slow or unreachable databases.apps/ai/src/agents/cofounder/curator.py-420-433 (1)
420-433: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winAudit fallback never reaches JSONL on runtime PostgreSQL failures
asyncio.get_running_loop()raises in the synchronous path, andpool.submit(asyncio.run, _pg_write)passes the coroutine function instead of a coroutine object. Since_file_audit_write()only runs onImportError, any non-import PostgreSQL failure drops the audit record instead of falling back.🤖 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 audit write fallback in curator.py only handles ImportError, so runtime PostgreSQL failures never reach _file_audit_write. Update the try/except around _pg_write so any PostgreSQL write failure from the async/sync path falls through to the JSONL fallback, and make sure the ThreadPoolExecutor branch in the asyncio.get_running_loop logic submits a coroutine object to asyncio.run rather than the coroutine function. Keep the fallback behavior centralized around _pg_write and _file_audit_write so non-import exceptions also trigger the JSONL write.apps/ai/src/session/mission_state.py-182-198 (1)
182-198: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
generate_briefflag silently gates whetherupdate_reason/changed_fieldsare persisted.
state.last_update_reasonandstate.last_changed_fieldsare only assigned whengenerate_brief=True. Any caller that wants explainability tracking without triggering brief auto-generation (generate_brief=False) will have itsupdate_reason/changed_fieldsarguments silently dropped, since they're never applied tostatebefore the INSERT. This conflates two independent concerns (brief generation vs. explainability bookkeeping) under one flag.🛠️ 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🤖 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 182 - 198, The update_mission_state function is incorrectly tying explainability fields to the generate_brief flag, so update_reason and changed_fields can be dropped when brief generation is disabled. Update the mission state assignment logic so MissionState.last_update_reason and MissionState.last_changed_fields are set independently of brief generation, and keep generate_brief only for the prepared_brief auto-generation path.apps/ai/src/session/brief_generator.py-18-49 (1)
18-49: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winSynchronous
chat_completioncall blocks the event loop.The comment at Line 30 acknowledges
chat_completionis synchronous, yet it's invoked directly inside anasync def. This is called fromupdate_mission_state(awaited on the state-write hot path), so every request that triggers brief generation will block the entire event loop for the duration of the LLM call, stalling all other concurrent coroutines on that worker.🛠️ Proposed fix
+ import asyncio + ... try: # chat_completion is synchronous - brief = chat_completion( + 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, )🤖 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 18 - 49, The synchronous chat_completion call inside generate_prepared_brief is blocking the event loop because it is executed directly within the async function on the update_mission_state hot path. Move the chat_completion invocation off the event loop by running it in a background thread or equivalent async wrapper, then keep the existing brief handling, persistence, and logging in generate_prepared_brief unchanged.
🟡 Minor comments (7)
.github/workflows/ci.yml-87-97 (1)
87-97: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winIntegration-tests job has no actual
if: falsegate.The comment and echoed text reference an
if: falsecondition that is never actually applied to the job — it will run unconditionally on every trigger. Currently harmless (echo-only), but contradicts the stated "disabled by default" design and will silently execute for real once Docker steps are filled in.🔧 Proposed fix
integration-tests: name: Integration Tests runs-on: ubuntu-latest needs: [unit-tests] - # Enable by adding: if: github.event_name == 'push' (or remove this comment) + if: false # Enable by removing this line (or changing to a real condition) steps:🤖 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 @.github/workflows/ci.yml around lines 87 - 97, The integration-tests job is missing the actual disable-by-default guard that the workflow text refers to, so add the intended job-level if condition on integration-tests itself rather than only documenting it in comments or echoed text. Update the integration-tests job definition in ci.yml so it is explicitly gated (for example via an if expression that evaluates to false by default), and keep the existing job name and steps unchanged otherwise..opencode/context/domain/architecture.md-153-160 (1)
153-160: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winUse the real tool filenames here.
This table still namespause_failed_payment_retry/flag_churn_risk_customer, but the registry modules arepause_payment_retry.py/flag_churn_risk.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/domain/architecture.md around lines 153 - 160, The tool registry table is using symbolic names that do not match the actual module filenames, so update the entries in the architecture document to use the real tool filenames instead of the old names. Fix the rows for pause_failed_payment_retry and flag_churn_risk_customer by aligning them with pause_payment_retry.py and flag_churn_risk.py, and verify the other tool names remain consistent with the registry modules referenced in this section.README.md-189-194 (1)
189-194: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winUse the actual module names here.
This table still listspause_failed_payment_retryandflag_churn_risk_customer, but the checked-in tools arepause_payment_retry.pyandflag_churn_risk.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 `@README.md` around lines 189 - 194, Update the tools table in README to use the actual checked-in module names instead of the old action names; replace `pause_failed_payment_retry` with `pause_payment_retry.py` and `flag_churn_risk_customer` with `flag_churn_risk.py`. Keep the rest of the table intact and make sure the entries still match the corresponding tool behavior and triggers..opencode/context/adr/001-sarthi-v4-architecture-evolution.md-164-168 (1)
164-168: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winAlign the tool names with the checked-in modules.
pause_failed_payment_retryandflag_churn_risk_customerdon't match the actual tool files (pause_payment_retry.py,flag_churn_risk.py), so this ADR will drift from the registry docs.🤖 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/adr/001-sarthi-v4-architecture-evolution.md around lines 164 - 168, The ADR tool list is using outdated tool names that no longer match the checked-in modules. Update the entries in the tools section to use the actual names from the registry and files, specifically align `pause_failed_payment_retry` with `pause_payment_retry` and `flag_churn_risk_customer` with `flag_churn_risk`, while keeping the other tool names consistent with the implementations in `apps/ai/src/agents/tools/`.apps/ai/src/agents/authority_manifest.py-113-117 (1)
113-117: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
can_execute_toolignores thecan_execute_toolsflag.The function only checks
tool_id in auth.allowed_tool_ids, never consultingauth.can_execute_tools. If any future/edited manifest entry setscan_execute_tools=Falsebut leaves a stale non-emptyallowed_tool_ids, this function would incorrectly authorize execution. Currently self-consistent by convention (Correlation Agent pairsFalsewith[]), but the gate should be enforced explicitly rather than relying on data discipline.🛡️ Proposed fix
def can_execute_tool(agent_name: str, tool_id: str) -> bool: auth = get_authority(agent_name) if auth is None: return False - return tool_id in auth.allowed_tool_ids + return auth.can_execute_tools and tool_id in auth.allowed_tool_ids🤖 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/authority_manifest.py` around lines 113 - 117, can_execute_tool currently authorizes purely from auth.allowed_tool_ids and ignores the auth.can_execute_tools gate. Update can_execute_tool in authority_manifest.py to first check the can_execute_tools flag from get_authority(agent_name) and return False when it is disabled, then only consult allowed_tool_ids when execution is permitted. Keep the fix localized to can_execute_tool and use auth as the source of truth for both checks.apps/core/internal/web/templates/partials/command_alert_lineage.html-28-36 (1)
28-36: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winMissing explicit "approve" tier styling.
Only
"auto"(green) and"review"(yellow) are handled; any other tier — including"approve", part of the HITL tier vocabulary used elsewhere in this codebase — falls into the catch-allelsebranch styled as red/danger, alongside truly unrecognized values. This misrepresents an "approve" tier's risk level to founders reviewing the lineage panel.🎨 Proposed fix
- <span class="text-[10px] px-2 py-0.5 rounded-full {{if eq .Tier "auto"}}bg-green-900/30 text-green-400{{else if eq .Tier "review"}}bg-yellow-900/30 text-yellow-400{{else}}bg-red-900/30 text-red-400{{end}}"> + <span class="text-[10px] px-2 py-0.5 rounded-full {{if eq .Tier "auto"}}bg-green-900/30 text-green-400{{else if eq .Tier "review"}}bg-yellow-900/30 text-yellow-400{{else if eq .Tier "approve"}}bg-blue-900/30 text-blue-400{{else}}bg-red-900/30 text-red-400{{end}}">🤖 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_alert_lineage.html` around lines 28 - 36, The suggested action badge rendering in command_alert_lineage.html is missing explicit handling for the "approve" tier, so it is currently treated as the red catch-all state. Update the conditional in the SuggestedActions template block to recognize .Tier == "approve" separately from the fallback branch, and style it with an appropriate non-danger color consistent with the existing tier vocabulary used alongside "auto" and "review".apps/core/internal/web/templates/partials/command_operating_layer.html-58-61 (1)
58-61: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winSplit Active Agent Roles into separate pills
ActiveAgentRolesis passed through as a single JSON string, so this template renders one badge with the literal array text instead of one pill per role. Parse it into[]stringin the handler andrangeover it 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 in command_operating_layer.html is rendering the raw JSON string as one badge instead of separate role pills. Update the handler that populates this template to parse ActiveAgentRoles into a []string, then change the template block around .ActiveAgentRoles to range over each role and render one pill per item. Use the existing ActiveAgentRoles field and the template block in command_operating_layer.html to locate the change.
🧹 Nitpick comments (10)
apps/core/internal/web/handler.go (1)
1627-1763: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract the shared SSE loop to remove 4× duplication.
APICommandMissionEvents,APICommandHITLEvents, andAPICommandSessionEventsare near-identical copies ofAPICommandChatEvents, differing only in the subscribed event type and theconnectedbanner text. Consider a single helper to keep them in sync (and to apply the cleanup fix above in one place).♻️ Sketch
func (h *Handler) streamTenantEvents(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 h.sseHub.Unsubscribe(tenantID, sub.ID) defer func() { recover() }() 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 APICommandMissionEvents, APICommandHITLEvents, and APICommandSessionEvents duplicate the same subscription/streaming loop, so extract the shared logic into a helper like streamTenantEvents on Handler and have each endpoint pass only the event type and connected banner text. Move the common setup, heartbeat loop, message forwarding, and cleanup into that helper so the three endpoints stay in sync and the unsubscribe cleanup is applied consistently in one place.apps/core/internal/web/sse_hub.go (1)
89-92: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winLog dropped events in the non-blocking broadcast.
When a subscriber's buffered channel is full, the
defaultbranch silently discards the event, giving no signal that SSE messages are being lost under load. The project's SSE convention calls for logging on the non-blocking drop path.♻️ Proposed change
select { case sub.Channel <- []byte(msg): default: + // buffer full — subscriber is slow; drop rather than block + log.Printf("sse: dropping %q event for tenant=%s sub=%s (buffer full)", event.Type, tenantID, sub.ID) }As per coding guidelines: "non-blocking broadcast via
tryBroadcast()withselect { case ch <- msg: default: log }".🤖 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 89 - 92, The non-blocking send in tryBroadcast on SSEHub currently drops messages silently when sub.Channel is full. Update the default branch of the select to log a dropped-event warning with enough context to identify the subscriber/message, while keeping the send non-blocking. Use the existing SSEHub and tryBroadcast symbols to locate the broadcast path and preserve the current behavior for successful sends.Source: Coding guidelines
docker-compose.llmops.yml (1)
5-125: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftSignificant duplication with
docker-compose.showcase.yml.The postgres, redis, minio, langfuse, and langfuse-worker service blocks here are nearly identical to their counterparts in
docker-compose.showcase.yml(differing only in container/volume names). Consider extracting shared service definitions into a base compose file included viadocker compose -f base.yml -f llmops.yml(orextends) to avoid drift as these services evolve.🤖 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 `@docker-compose.llmops.yml` around lines 5 - 125, The service definitions for postgres, redis, minio, langfuse, and langfuse-worker are duplicated across compose files, so update the compose setup to share common configuration instead of maintaining near-identical blocks. Refactor these services in docker-compose.llmops.yml to reuse a base compose definition (for example via docker compose -f base.yml -f llmops.yml or extends), keeping only environment-specific overrides like container_name, volume, and port differences. Use the service names postgres, redis, minio, langfuse, and langfuse-worker to locate the duplicated sections and align them with the shared source of truth..opencode/context/standards/coding-standards.md (1)
189-213: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMake the example match the logging standard above.
The surrounding standards requirestructlog, but this snippet still uses stdlibloggingand positional formatting.Proposed fix
- import logging + import structlog from typing import Any - log = logging.getLogger(__name__) + log = structlog.get_logger(__name__) - log.info("my_tool_name %s — tier=%s", tenant_id, tool_def["hitl_tier"]) + log.info("my_tool_name", tenant_id=tenant_id, tier=tool_def["hitl_tier"])🤖 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 189 - 213, The example in the tool template still uses stdlib logging and positional formatting, which conflicts with the surrounding logging standard that requires structlog. Update the example around tool_def and execute to use structlog-style logging, with a bound logger and key/value event logging instead of logging.getLogger(__name__) and log.info(... %s ...). Keep the example aligned with the existing identifiers execute, tool_def, and log so it demonstrates the standard correctly.apps/ai/infrastructure/migrations/004_control_plane_audit.sql (1)
11-12: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low valueConsider CHECK constraints for enum-like columns.
approval_stateandoutcomeare documented as closed sets (auto | review | approve | blocked,completed | blocked | failed) but stored as unconstrainedVARCHAR. ACHECKconstraint would catch bad writes at the DB layer.🤖 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 11 - 12, The migration for the audit table currently stores `approval_state` and `outcome` as unconstrained `VARCHAR` columns even though they represent closed sets. Update the `004_control_plane_audit.sql` schema to add DB-level `CHECK` constraints on these columns so only the documented values are accepted. Use the column names `approval_state` and `outcome` in the existing migration definition to keep the fix localized.apps/ai/src/risk/output_risk.py (1)
110-112: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
_severity_rankis duplicated verbatim inprompt_risk.py.Consider moving this helper (and possibly the rule-description pattern) into
src/risk/__init__.pyor a shared module to avoid drift between the two scanners.🤖 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/risk/output_risk.py` around lines 110 - 112, The _severity_rank helper is duplicated in output_risk.py and prompt_risk.py, so consolidate it into a shared location to prevent drift. Move _severity_rank (and any reusable rule-description pattern it depends on) into src/risk/__init__.py or another shared risk module, then update both scanners to import and use the shared helper instead of keeping separate copies.apps/ai/src/control_plane/audit.py (1)
34-38: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider a shared connection pool instead of a connect-per-call pattern.
Every
log_eventcall without a caller-suppliedconnopens and tears down a brand-newasyncpgconnection. Given this runs on every gated tool execution, a sharedasyncpg.Pool(e.g. injected at app startup) would avoid repeated connection-setup overhead and reduce risk of exhausting DB connections under load.🤖 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 34 - 38, The log_event flow currently creates a fresh asyncpg connection whenever conn is not provided, which causes connect-per-call overhead and can exhaust database connections under load. Update the audit path around log_event to use a shared asyncpg.Pool instead of calling asyncpg.connect inline, and have the pool injected or otherwise available at startup. Keep support for an explicitly supplied conn, but route the default path through the shared pool and adjust the cleanup logic so the pool-managed connection lifecycle is handled correctly.apps/ai/src/agents/tools/schedule_customer_checkin.py (1)
46-56: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid blocking Slack I/O in async paths
execute()callsSlackClient().client.chat_postMessage(...)on the event loop. Switch this toAsyncWebClient/await, or offload the send toasyncio.to_thread(...). The same pattern also appears inapps/ai/src/integrations/slack_client.py::SlackClient.send_guardian_alert.🤖 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 - 56, The Slack send in `schedule_customer_checkin.py::execute` is performing blocking I/O on the async path by calling `SlackClient().client.chat_postMessage(...)` directly. Update this flow to use a non-blocking approach, either by switching the underlying client usage to `AsyncWebClient` and awaiting the message send, or by offloading the existing send call with `asyncio.to_thread(...)`; make the same change in `SlackClient.send_guardian_alert` so both Slack message paths avoid blocking the event loop.apps/core/internal/web/command_center_test.go (1)
525-550: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider adding a DB-backed case to exercise the data-population path.
All operating-layer tests use
NewHandler(nil, nil), so theif h.db != nilbranch inAPICommandOperatingLayer(which formatsPendingDecisions/ActiveAgentRolesfrom the DB) is never exercised — only static headings are asserted. A test with a stubbed/sqlmock DB row would catch formatting regressions like the raw-JSON-array display issue flagged incommand_operating_layer.html.🤖 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/command_center_test.go` around lines 525 - 550, The current `TestAPICommandOperatingLayer_ReturnsValidHTML` only covers the no-DB path because it uses `NewHandler(nil, nil)`, so `APICommandOperatingLayer`’s `h.db != nil` data-population branch is never exercised. Add a DB-backed test case for `APICommandOperatingLayer` using a stubbed or sqlmock database row, and assert the rendered `PendingDecisions` and `ActiveAgentRoles` formatting in the HTML so regressions in the DB-to-view mapping are caught.apps/core/hitl_queue_test.go (1)
25-25: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueStale comment contradicts passing tests.
// NEW: HTMX screens - will fail until implementedis misleading now that these routes are wired to working handlers and the tests below assert on their real output. Worth removing/updating to avoid confusing future readers.🤖 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/hitl_queue_test.go` at line 25, The inline comment in the hitl queue test is now stale and contradicts the implemented HTMX routes and passing assertions. Remove or rewrite the comment near the test setup in hitl_queue_test.go so it accurately reflects the current behavior of the relevant test cases and does not suggest the screens are unimplemented.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: ae7808f3-9850-4271-9537-782ba982e77e
⛔ Files ignored due to path filters (1)
apps/ai/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (51)
.githooks/pre-push.github/workflows/ci.yml.gitignore.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.mdMakefileREADME.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/hubspot.pyapps/ai/src/integrations/quickbooks.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_agentic_comprehensive.pyapps/ai/tests/unit/test_control_plane.pyapps/ai/tests/unit/test_qa_agent.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.htmldocker-compose.dev.ymldocker-compose.llmops.ymldocker-compose.showcase.yml
💤 Files with no reviewable changes (1)
- apps/ai/src/integrations/hubspot.py
| "reason": "Output risk scan blocked draft", | ||
| }, | ||
| ) | ||
| _AUDIT.log(audit_event) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== draft_investor_update.py ==\n'
file='apps/ai/src/agents/tools/draft_investor_update.py'
wc -l "$file"
sed -n '1,220p' "$file"
printf '\n== audit logger definition ==\n'
audit='apps/ai/src/control_plane/audit.py'
wc -l "$audit"
sed -n '1,220p' "$audit"
printf '\n== search for AuditLogger usage and log/log_event methods ==\n'
rg -n "class AuditLogger|def log_event|def log\(|_AUDIT\.log\(|_AUDIT\.log_event\(" apps/ai/srcRepository: Aparnap2/Track_Guard
Length of output: 7341
Use await _AUDIT.log_event(...) in both audit paths
AuditLogger only defines async log_event(...), so _AUDIT.log(audit_event) raises at runtime and stops execute() on both the blocked and successful paths. Replace both calls with await _AUDIT.log_event(audit_event).
🤖 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 97,
`DraftInvestorUpdate.execute` is calling the audit logger with the wrong API,
which will fail at runtime because `AuditLogger` only exposes `log_event`.
Update both audit paths in `draft_investor_update.py` so they await
`_AUDIT.log_event(audit_event)` instead of calling `_AUDIT.log(audit_event)`,
and make sure the blocked and successful execution branches both use the async
method consistently.
| class AuditLogger: | ||
| """Logs agent actions to the audit_log table for inspection.""" | ||
|
|
||
| async def log_event( | ||
| self, | ||
| event: AuditEvent, | ||
| conn: asyncpg.Connection | None = None, | ||
| ) -> bool: |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm whether tool files call AuditLogger.log() (missing method) instead of log_event().
rg -nP '_AUDIT\.(log|log_event)\(' apps/ai/src
rg -nP 'class AuditLogger' -A5 apps/ai/src/control_plane/audit.pyRepository: Aparnap2/Track_Guard
Length of output: 497
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- audit.py outline ---'
ast-grep outline apps/ai/src/control_plane/audit.py --view expanded || true
echo '--- audit.py relevant lines ---'
sed -n '1,220p' apps/ai/src/control_plane/audit.py | cat -n
echo '--- draft_investor_update.py relevant lines ---'
sed -n '1,220p' apps/ai/src/agents/tools/draft_investor_update.py | cat -n
echo '--- search for AuditLogger usage / aliases ---'
rg -n 'AuditLogger\(|\.log_event\(|\.log\(|__getattr__|def log\(' apps/ai/srcRepository: Aparnap2/Track_Guard
Length of output: 8842
Call await _AUDIT.log_event(...) here apps/ai/src/agents/tools/draft_investor_update.py:97,120 uses _AUDIT.log(audit_event), but AuditLogger only exposes log_event(...). Those calls will raise AttributeError and abort the tool flow.
🤖 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 17 - 24, The audit call
sites are using a non-existent AuditLogger method, which will raise
AttributeError during the tool flow. Update the references in
draft_investor_update to call await _AUDIT.log_event(...) instead of
_AUDIT.log(...), matching the public API exposed by AuditLogger.log_event and
preserving the existing AuditEvent payload.
| state.prepared_brief, | ||
| state.pending_decisions, | ||
| state.last_updated_by, | ||
| state.last_update_reason, | ||
| json.dumps(state.last_changed_fields) if state.last_changed_fields else '[]', | ||
| json.dumps(state.active_agent_roles) if state.active_agent_roles else '[]', | ||
| json.dumps(state.policy_state) if state.policy_state else None, | ||
| ) | ||
| await conn.close() | ||
| if generate_brief and not state.prepared_brief: | ||
| try: | ||
| from src.session.brief_generator import generate_prepared_brief | ||
| await generate_prepared_brief(state.tenant_id) | ||
| except Exception: | ||
| log.exception("generate_prepared_brief callback failed") |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the relevant file around the cited lines
sed -n '240,320p' apps/ai/src/session/mission_state.py
# Find all references to pending_decisions in the session code
rg -n "pending_decisions|jsonb|asyncpg|prepare|execute\(" apps/ai/src/session -S
# Show the definition of MissionState if it's in this file or nearby
rg -n "class MissionState|pending_decisions" apps/ai/src/session -SRepository: Aparnap2/Track_Guard
Length of output: 6275
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the session schema/migration that defines pending_decisions
rg -n "pending_decisions|prepared_brief|last_changed_fields|active_agent_roles|policy_state" . -g '*.sql' -g '*.py' -S
# Show the MissionState model and write/load paths in full context
sed -n '1,220p' apps/ai/src/session/mission_state.py
sed -n '1,180p' apps/ai/src/session/context.pyRepository: Aparnap2/Track_Guard
Length of output: 18339
🌐 Web query:
asyncpg JSONB Python list dict parameter adaptation default jsonb encode decode
💡 Result:
In asyncpg, JSONB values are not automatically encoded or decoded by default [1]. If you query a JSONB column, it will be returned as a string (the raw text representation from PostgreSQL) [1][2]. To enable automatic conversion between Python lists/dicts and PostgreSQL JSONB, you must explicitly register a custom type codec using the connection.set_type_codec() method [3][2]. ### Registering Custom JSONB Codecs You can configure your connection to automatically handle JSON serialization and deserialization [3]. It is recommended to apply this configuration using the init parameter when creating a connection pool to ensure all connections inherit the behavior [4]. python import json import asyncpg # Define custom encoder and decoder def encoder(value): return json.dumps(value) def decoder(value): return json.loads(value) # Register the codec async def setup_connection(conn): await conn.set_type_codec( 'jsonb', encoder=encoder, decoder=decoder, schema='pg_catalog', format='text' # Use 'text' format for JSON/JSONB ) # Use with a connection pool pool = await asyncpg.create_pool(dsn, init=setup_connection) ### Important Considerations 1. Automatic Encoding Risks: When you register a custom encoder, asyncpg will attempt to use it for any Python object it doesn't recognize as a standard type [5]. If you pass a list of dicts to a query, asyncpg may treat the structure as a PostgreSQL array rather than a single JSONB document [5]. To avoid ambiguous behavior, always explicitly cast your parameters in SQL, such as $1::jsonb [6]. 2. Codec Format: When setting the codec for JSON/JSONB, use format='text' [3][7]. Attempting to use binary format without custom handling of the PostgreSQL binary JSONB layout will result in errors [8][4]. 3. Global vs. Local: Codecs are registered per-connection [3]. Using the init argument in create_pool is the standard way to apply this change globally across your application [4].
Citations:
- 1: In Record complex data coming as string MagicStack/asyncpg#560
- 2: https://stackoverflow.com/questions/79727814/pythonasyncpgpostgresql-returns-jsonb-column-as-a-string-instead-of-an-object
- 3: https://magicstack.github.io/asyncpg/devel/usage.html
- 4: Custom Type Conversions issues MagicStack/asyncpg#148
- 5: sql queries using unnest with jsonb MagicStack/asyncpg#345
- 6: JSONB positional parameter MagicStack/asyncpg#912
- 7: https://magicstack.github.io/asyncpg/devel/api/index.html
- 8: Error when trying to set JSONB as a custom type codec MagicStack/asyncpg#140
🌐 Web query:
asyncpg usage JSONB Python dict list parameter result decoding default docs
💡 Result:
In asyncpg, JSON and JSONB types are not automatically encoded from or decoded to Python dictionaries or lists by default [1][2]. When you query a JSONB column, it is returned as a string, and when you attempt to pass a Python dict or list directly as a parameter, it may fail unless you have configured a custom type codec [3][1][4]. To enable automatic conversion, you must register a custom type codec using the Connection.set_type_codec method [3][4]. Configuring Automatic Conversion You can configure asyncpg to automatically use the json module to handle encoding and decoding [3]: import json import asyncpg async def setup_json_codec(conn): await conn.set_type_codec( 'json', encoder=json.dumps, decoder=json.loads, schema='pg_catalog') # Register the same for jsonb await conn.set_type_codec( 'jsonb', encoder=json.dumps, decoder=json.loads, schema='pg_catalog') # When using a connection pool, use the init parameter to ensure # the codec is set on all connections created by the pool pool = await asyncpg.create_pool(dsn, init=setup_json_codec) Key Considerations 1. Persistence: When using a connection pool, always use the init argument in create_pool to register codecs [4]. This ensures the codec is applied to every connection in the pool, rather than attempting to set it on an individual connection after acquisition [4]. 2. Manual Encoding: If you choose not to register a codec, you must manually serialize Python objects to strings (e.g., using json.dumps) before passing them as parameters to your SQL queries, and manually deserialize the returned strings (e.g., using json.loads) after fetching rows [5][4]. 3. Complexity with Unnest: Be aware that automatic JSON encoding can sometimes lead to unexpected behavior in complex queries, such as when using unnest with arrays of JSONB objects, as asyncpg may incorrectly interpret the structure of the data [5]. In such cases, manually serializing the data to JSON strings and casting them in the SQL query (e.g., $1::jsonb[]) is often safer [5].
Citations:
- 1: JSONB is not automatically decoded MagicStack/asyncpg#273
- 2: In Record complex data coming as string MagicStack/asyncpg#560
- 3: https://magicstack.github.io/asyncpg/devel/usage.html
- 4: json_agg and json_build_object raw queries return string type rather than json MagicStack/asyncpg#432
- 5: sql queries using unnest with jsonb MagicStack/asyncpg#345
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Look for any asyncpg type codec registration or connection wrapper
rg -n "set_type_codec|create_pool|jsonb|json.dumps|json.loads|register.*codec|init=.*asyncpg|asyncpg.connect" apps/ai/src -S
# Inspect database config utilities that may wrap asyncpg connections
fd -a "database.py|*.py" apps/ai/src/config apps/ai/src -t fRepository: Aparnap2/Track_Guard
Length of output: 8185
🌐 Web query:
site:magicstack.github.io asyncpg jsonb default codec encode decode list dict parameter docs
💡 Result:
By default, asyncpg does not automatically handle JSON or JSONB types as Python lists or dictionaries. These types are encoded and decoded as text by default [1]. To automatically convert PostgreSQL JSON/JSONB types to and from Python objects (like lists and dictionaries), you must register a custom codec using the connection.set_type_codec method [1]. The standard approach is to use the Python json module for encoding and decoding [1]. Below is the recommended implementation: import asyncpg import json async def main: conn = await asyncpg.connect # Register the codec for 'json' (or 'jsonb') await conn.set_type_codec( 'json', encoder=json.dumps, decoder=json.loads, schema='pg_catalog') # Now, passing a dict or list will work automatically data = {'key': 'value', 'list': [1][2][3]} res = await conn.fetchval('SELECT $1::json', data) print(res) await conn.close Key details for setting up the codec: - Type Name: Use 'json' or 'jsonb' as the typename [1]. - Schema: For JSON/JSONB, the schema is typically 'pg_catalog' [1]. - Encoder/Decoder: Provide callable functions like json.dumps and json.loads [1]. - Format: The default format is 'text', which is appropriate for JSON encoding [1][4]. For more information, refer to the Custom Type Conversions section of the official asyncpg documentation [1].
Citations:
- 1: https://magicstack.github.io/asyncpg/devel/usage.html
- 2: https://magicstack.github.io/asyncpg/current/usage.html
- 3: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html
- 4: https://magicstack.github.io/asyncpg/devel/api/index.html
🌐 Web query:
asyncpg jsonb parameter encoding dict list default behavior docs issue
💡 Result:
In asyncpg, the default behavior for json and jsonb types is to treat them as plain text (strings), meaning the library does not automatically encode Python dictionaries or lists into JSON strings, nor does it decode them back into Python objects upon retrieval [1][2][3][4]. If you pass a dictionary or list directly to a query parameter without a custom codec, asyncpg will likely raise a type error because it expects a string or cannot infer the correct binary representation for a complex Python object [5]. To enable seamless handling of JSON data, you must configure a custom type codec using the connection.set_type_codec method [1][3]. Recommended Setup: To handle JSONB (or JSON) automatically, you can register a codec that uses the standard Python json module to encode and decode values in text format [1][3]: import json await conn.set_type_codec( 'jsonb', encoder=json.dumps, decoder=json.loads, schema='pg_catalog') Key Implementation Details: 1. Encoding/Decoding: By using the 'text' format (the default when format is omitted), you avoid the complexity of managing PostgreSQL's internal JSONB binary format (which includes a version header byte) [6][7]. 2. Explicit Casting: It is best practice to explicitly cast your parameters in SQL (e.g., $1::jsonb) to avoid type inference ambiguities, especially when using positional parameters [8][5]. 3. Avoiding Double Encoding: If you have configured a custom codec, ensure you pass the raw Python object (dict/list) to the query. If you manually encode the object to a JSON string before passing it, the codec will encode it again, resulting in a double-encoded string [8]. If you encounter issues where your data is still returned as a string, confirm that set_type_codec was successfully called on your connection or pool object before executing the query [2].
Citations:
- 1: https://github.com/MagicStack/asyncpg/blob/master/docs/usage.rst
- 2: In Record complex data coming as string MagicStack/asyncpg#560
- 3: https://magicstack.github.io/asyncpg/devel/usage.html
- 4: http://docs.sqlalchemy.org/en/latest/dialects/postgresql.html
- 5: what is the correct usage of fetch api with jsonb #> operator ? MagicStack/asyncpg#449
- 6: Error when trying to set JSONB as a custom type codec MagicStack/asyncpg#140
- 7: no binary format encoder for type jsonb MagicStack/asyncpg#1144
- 8: JSONB positional parameter MagicStack/asyncpg#912
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect other asyncpg jsonb inserts in the repo for the established pattern
sed -n '1,220p' apps/ai/src/control_plane/audit.py
sed -n '1,220p' apps/ai/src/services/trust_battery_db.py
# Search for direct Python dict/list bindings to ::jsonb
rg -n "::jsonb|jsonb" apps/ai/src -S | sed -n '1,120p'Repository: Aparnap2/Track_Guard
Length of output: 11606
Serialize pending_decisions before writing to mission_states
state.pending_decisions is passed to asyncpg as a raw Python list, while the other JSON/JSONB values are serialized first. This insert should json.dumps(...) it too so update_mission_state doesn’t fail when pending_decisions is populated.
🐛 Proposed fix
state.prepared_brief,
- state.pending_decisions,
+ json.dumps(state.pending_decisions) if state.pending_decisions else None,
state.last_updated_by,📝 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.prepared_brief, | |
| state.pending_decisions, | |
| state.last_updated_by, | |
| state.last_update_reason, | |
| json.dumps(state.last_changed_fields) if state.last_changed_fields else '[]', | |
| json.dumps(state.active_agent_roles) if state.active_agent_roles else '[]', | |
| json.dumps(state.policy_state) if state.policy_state else None, | |
| ) | |
| await conn.close() | |
| if generate_brief and not state.prepared_brief: | |
| try: | |
| from src.session.brief_generator import generate_prepared_brief | |
| await generate_prepared_brief(state.tenant_id) | |
| except Exception: | |
| log.exception("generate_prepared_brief callback failed") | |
| state.prepared_brief, | |
| json.dumps(state.pending_decisions) if state.pending_decisions else None, | |
| state.last_updated_by, | |
| state.last_update_reason, | |
| json.dumps(state.last_changed_fields) if state.last_changed_fields else '[]', | |
| json.dumps(state.active_agent_roles) if state.active_agent_roles else '[]', | |
| json.dumps(state.policy_state) if state.policy_state else None, |
🧰 Tools
🪛 ast-grep (0.44.1)
[info] 280-280: use jsonify instead of json.dumps for JSON output
Context: json.dumps(state.last_changed_fields)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 281-281: use jsonify instead of json.dumps for JSON output
Context: json.dumps(state.active_agent_roles)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 282-282: use jsonify instead of json.dumps for JSON output
Context: json.dumps(state.policy_state)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
🤖 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 - 291, Serialize
state.pending_decisions before the mission_states write in update_mission_state,
matching the existing handling for last_changed_fields, active_agent_roles, and
policy_state. The asyncpg insert currently passes pending_decisions as a raw
Python list, so change the value assembly in mission_state.py to
json.dumps(state.pending_decisions) when present (or '[]' when empty) to ensure
the prepared statement accepts it without failing.
1 participant
Summary
Tiered runtime profiles for 16GB laptop constraints and local-first CI gates.
Changes
Tiered Runtime Profiles
sarthi-dev-*,sarthi-llmops-*,sarthi-showcase-*)iterateswarm-netfor service discovery compatibilityLocal CI Infrastructure
Bugfixes (pre-existing)
exceptinhubspot.py(SyntaxError)try:block inquickbooks.py(SyntaxError)lineagefield intest_agentic_comprehensive.py(Pydantic schema update)test_qa_agent.py(false positive)DevOps Hygiene
Summary by CodeRabbit
New Features
Bug Fixes