diff --git a/.opencode/context/adr/001-sarthi-v4-architecture-evolution.md b/.opencode/context/adr/001-sarthi-v4-architecture-evolution.md index 1759111..62d284c 100644 --- a/.opencode/context/adr/001-sarthi-v4-architecture-evolution.md +++ b/.opencode/context/adr/001-sarthi-v4-architecture-evolution.md @@ -23,7 +23,7 @@ The IterateSwarm Sarthi agent platform underwent a major architecture evolution ## Decision -We made seven architectural decisions to address these issues. Each is documented below. +We made ten architectural decisions to address these issues. Each is documented below. ### Decision 1: HTMX SSE over WebSocket / Raw JS EventSource @@ -121,6 +121,84 @@ var specialistRoutes = map[string]specialistRoute{ **Rationale**: The removed types have real implementations in `apps/ai/src/` (Python) or `apps/core/internal/agents/` (Go). Keeping stubs created confusion — developers had to check whether a type was a stub or the real implementation. Removing them means "if it's in stubs.go, it's a real convenience type used here." This reduces cognitive load and prevents stale stubs from diverging from real implementations. +### Decision 8: SSEHub Event-Type Filtering (2026-06-28) + +**What**: Introduced a dedicated SSEHub with event-type filtering to replace the raw `chatBroadcast chan fiber.Map` for chat-only SSE. Event filtering enables domain-specific SSE streams (chat, mission state, HITL, session events). + +**How**: +```go +type SSEEvent struct { + Type string `json:"type"` + Payload string `json:"payload"` +} + +type Subscription struct { + ID string + Channel chan []byte + Types []string // empty = all events +} +``` +- `SSEHub.Subscribe(tenantID, "chat")` returns a subscription filtered to chat events only. +- `SSEHub.Broadcast(tenantID, SSEEvent{Type: "chat", ...})` only delivers to subscribers whose `Types` filter matches (or whose filter is empty = all events). +- Non-blocking fan-out with `select/default` on per-subscriber channels (buffered to 64). +- Backward-compatible: `tryBroadcast()` also pushes to `sseHub.Broadcast("default", SSEEvent{Type: "chat", ...})` alongside the legacy `chatBroadcast` channel. +- The `Subscription.Types` filter eliminates the need for separate SSE channels per event domain — one hub, typed routing. + +**Rationale**: The original `chatBroadcast` channel was chat-only. As the SSE surface grows to include mission state updates, HITL approval signals, and session events, a single channel with event-type filtering prevents channel explosion (one channel per event type) while keeping the fan-out pattern simple. Per-subscriber channels (buffered 64) replace the single shared buffer (100), reducing head-of-line blocking. + +### Decision 9: Tool Calling Surface — ToolRegistry + HITL Tier Mapping (2026-06-28) + +**What**: Created a tool calling surface where agent actions are defined as standalone `ToolDef` entries, registered in a global `TOOL_REGISTRY`, and wired to the HITL manager for automatic tier assignment. + +**How**: +```python +# apps/ai/src/agents/tools/__init__.py +@dataclass +class ToolDef: + name: str + description: str + hitl_tier: str # "auto" | "review" | "approve" | "blocked" + fn: Callable[..., Awaitable[dict]] + trigger_patterns: list[str] # e.g., ["FG-05", "BG-04"] +``` +- 4 tool implementations in `apps/ai/src/agents/tools/`: + - `pause_failed_payment_retry` — tier: review, triggered by FG-05 + - `draft_investor_update` — tier: approve, triggered by scheduled/manual + - `schedule_customer_checkin` — tier: auto, triggered by FG-03, BG-04 + - `flag_churn_risk_customer` — tier: auto, triggered by BG-06, BG-04 +- Tools auto-register on import via `register_tool(ToolDef(...))` at module bottom. +- `get_tools_for_tier(tier)` returns tools matching a HITL routing decision. +- `get_tools_for_patterns(pattern_ids)` returns tools matching triggered alert patterns. +- HITL tier strings are shared with `HITLManager.route()` output: `"auto" | "review" | "approve" | "blocked"`. + +**Rationale**: Previously there was no explicit connection between alert patterns (FG-05, BG-04) and executable actions. Tools lived inside agent graphs with no centralized registry for discovery. The ToolDef pattern makes each tool self-documenting (name, description, tier, trigger patterns), and the global registry enables the HITL manager to suggest tools based on fired patterns. Each tool's `hitl_tier` is a first-class field, so routing decisions propagate from the HITL manager to tool execution without extra branching. + +### Decision 10: Slack Consolidation — SocketMode Extension, Not Bolt (2026-06-28) + +**What**: Extended the existing `SlackClient` with SocketMode listener support (`SocketModeClient` from `slack_sdk.socket_mode`) instead of adopting Slack's Bolt framework. Wired ACE (Acknowledge-Consequence-Evaluate) reflector loop into `slack_buttons.py` for feedback scoring. + +**How**: +```python +# apps/ai/src/integrations/slack_client.py +from slack_sdk import WebClient +from slack_sdk.socket_mode import SocketModeClient + +class SlackClient: + def __init__(self): + self.client = WebClient(token=os.getenv("SLACK_BOT_TOKEN")) + self.socket_client = SocketModeClient( + app_token=os.getenv("SLACK_APP_TOKEN"), + web_client=self.client + ) +``` +- SocketModeClient receives interactive payloads via Slack's Socket Mode (WebSocket), avoiding public HTTPS endpoints for Slack events. +- `slack_buttons.py` routes five button actions: `acknowledge`, `dispute`, `show_breakdown`, `log_decision`. +- Buttons wire to ACE reflector loop via `_send_feedback_signal()` → `score_from_button()` (Reflector) + `update_strategy_confidence()` (Curator). +- Button signals (`+1.0` for acknowledge, `-1.0` for dispute) update the trust battery and strategy confidence scores. +- Decision logging modal (`open_decision_modal`) captures structured decision data (decision, alternatives, reasoning) and persists via `log_decision` activity. + +**Rationale**: Adding SocketMode to the existing `SlackClient` avoids the Bolt framework dependency and its opinionated middleware stack. The Socket Mode WebSocket connection eliminates the need for a public Slack Events API endpoint, simplifying deployment (no ngrok, no public URL for dev). The ACE loop wiring in `slack_buttons.py` closes the feedback loop: user button clicks → Reflector scores → strategy confidence updates — all without leaving the existing `slack_client.py` class. + --- ## Consequences @@ -136,6 +214,9 @@ var specialistRoutes = map[string]specialistRoute{ | Server-rendered bubbles | Single source of truth for HTML. XSS-safe via html.EscapeString. No client templates. | | MissionState POST | Clear write path. Dashboard is a dumb renderer. No client state hydration. | | Remove stubs | 40 fewer lines of stale dead code. Clear signal: "if it's here, it's real." | +| SSEHub event filtering | Event-type routing eliminates channel explosion. Per-subscriber channels reduce head-of-line blocking. | +| ToolRegistry + HITL mapping | Self-documenting tools with explicit tier. Central registry enables pattern-driven tool suggestion. | +| Slack SocketMode | No Bolt dependency. No public Slack Events URL needed. ACE loop closes feedback via reflector. | ### Tradeoffs @@ -148,6 +229,9 @@ var specialistRoutes = map[string]specialistRoute{ | Server-rendered bubbles | Harder to add interactive elements (copy buttons, actions) — would need JS event delegation. | | MissionState POST | Tight coupling between Python worker and Go schema. Schema changes require coordinated deploys. | | Remove stubs | `DiscordApprovalInput` still exists as a convenience type — could become stale if Discord integration changes. | +| SSEHub event filtering | Per-subscriber channels (64 buffer) per client increase memory under high concurrent connections. | +| ToolRegistry | Tools are auto-registered on import — import order dependencies could cause registration race at startup. | +| Slack SocketMode | Requires `SLACK_APP_TOKEN` (Socket Mode token) in addition to `SLACK_BOT_TOKEN`. WebSocket dependency for event ingestion. | ### Risks @@ -155,6 +239,8 @@ var specialistRoutes = map[string]specialistRoute{ 2. **Goroutine lifetime** — Long-lived goroutines (5-min timeout) hold references to request context. If the client disconnects, the goroutine continues until timeout or completion. Mitigation: check `ctx.Done()` in the goroutine loop pattern for future streaming responses. 3. **Specialist routing fragility** — Workflow type strings bypass compile-time checking. Mitigation: add a test that iterates all routes and verifies workflow names match registered Temporal workflow types. 4. **Signal ID mismatch** — The workflow ID used in the approval action must match the workflow ID that created the `planned_actions` row. If they diverge, signals are lost. Mitigation: store `workflow_id` in the `planned_actions` table and use it for signaling. +5. **Tool tier mismatch** — A tool's `hitl_tier` could diverge from the HITL manager's routing decision for a given pattern. Mitigation: `get_tools_for_tier()` provides a single source of truth — tools should be registered with tiers that match the HITL manager output. +6. **Slack SocketMode reconnection** — If the Socket Mode WebSocket drops, Slack button interactions are lost until reconnection. Mitigation: SocketModeClient auto-reconnects; monitor reconnection events via the `on_socket_connected` callback. --- diff --git a/.opencode/context/domain/architecture.md b/.opencode/context/domain/architecture.md index 863e03e..52ded95 100644 --- a/.opencode/context/domain/architecture.md +++ b/.opencode/context/domain/architecture.md @@ -62,7 +62,7 @@ └─────────────────────────────────────────────────────────────────┘ ``` -> **Architecture decisions documented in:** [ADR-001: Sarthi v4.0 Architecture Evolution](../adr/001-sarthi-v4-architecture-evolution.md) +> **Architecture decisions documented in:** [ADR-001: Sarthi v4.0 Architecture Evolution](../adr/001-sarthi-v4-architecture-evolution.md) (10 decisions, incl. SSEHub, ToolRegistry, Slack SocketMode) ## Key Design Decisions @@ -70,7 +70,12 @@ - **Framework**: Fiber v2 (`github.com/gofiber/fiber/v2`) - **HTMX**: Server-rendered HTML partials with `hx-trigger`, `hx-swap`, `hx-target` -- **SSE**: Fiber v2 `SetBodyStreamWriter` + `*bufio.Writer` for streaming (see [sse.go](/apps/core/internal/web/sse.go)) +- **SSE**: Fiber v2 `SetBodyStreamWriter` + `*bufio.Writer` for streaming + - **SSEHub** (`sse_hub.go`): Event-type filtered fan-out hub with per-subscriber channels + - `Subscribe(tenantID, eventTypes...)` creates typed subscriptions + - `Broadcast(tenantID, SSEEvent)` delivers only to matching subscribers (non-blocking) + - Backward-compatible: `tryBroadcast()` pushes to both legacy `chatBroadcast` channel and SSEHub + - Legacy polling-based SSE in [sse.go](/apps/core/internal/web/sse.go) (deprecated for new endpoints) - **Templates**: Embedded via `//go:embed` in `internal/web/templates/` - **Database**: Direct `database/sql` + `lib/pq` (not sqlc for command center — raw SQL queries) - **Temporal SDK**: `go.temporal.io/sdk` v1.39.0 for workflow dispatch and signaling @@ -120,10 +125,54 @@ ### Mission State System -- **Python AI layer** writes compiled operational state to `mission_state` table -- **Go handlers** read `mission_state` for dashboard KPIs, signals, health score -- Fields: mrr, burn_rate, runway_days, burn_alert, burn_severity, trust_score, churn_rate, error_spike, active_alerts, founder_focus +- **Python AI layer** writes compiled operational state to `mission_states` table (migration 004 reconciled schema drift `mission_state` → `mission_states`) +- **Go handlers** read `mission_states` for dashboard KPIs, signals, health score +- Core fields: mrr, burn_rate, runway_days, burn_alert, burn_severity, trust_score, churn_rate, error_spike, active_alerts, founder_focus +- **New cognitive offloading fields (2026-06-28):** + - `prepared_brief` — LLM-generated brief prepopulated for founder context before a decision + - `pending_decisions` — JSONB array of open decisions awaiting founder action + - `last_updated_by` — Which agent/specialist last wrote to MissionState (traceability) - Context budget: 800 tokens max for mission state compilation +- Dataclass in: `apps/ai/src/session/mission_state.py` +- Table references: `mission_states` (plural, not `mission_state`) + +### Tool Calling Surface (ToolRegistry + HITL Tier Mapping) + +New in 2026-06-28. Agent actions are defined as standalone `ToolDef` entries in a global `TOOL_REGISTRY` and wired to the HITL manager for automatic tier assignment. + +``` +ToolDef { + name: str # snake_case unique tool name + description: str # human-readable + hitl_tier: str # "auto" | "review" | "approve" | "blocked" + fn: Callable # async execute function + trigger_patterns: list # alert patterns that suggest this tool +} +``` + +**4 registered tools:** + +| Tool | Tier | Pattern | Description | +|------|------|---------|-------------| +| `pause_failed_payment_retry` | review | FG-05 | Pause Stripe retry on 3+ failed payments | +| `draft_investor_update` | approve | schedule/manual | Draft investor email for approval | +| `schedule_customer_checkin` | auto | FG-03, BG-04 | Auto-schedule at-risk customer reminder | +| `flag_churn_risk_customer` | auto | BG-06, BG-04 | Flag segment for churn monitoring | + +- Tools auto-register via `register_tool(ToolDef(...))` on module import +- `get_tools_for_tier(tier)` returns tools matching a routing decision +- `get_tools_for_patterns(pattern_ids)` returns tools matching triggered alerts +- Location: `apps/ai/src/agents/tools/__init__.py` + 4 tool modules + +### Slack Consolidation — SocketMode + ACE Loop + +New in 2026-06-28. The `SlackClient` was extended with `SocketModeClient` (WebSocket-based event ingestion) to handle interactive button payloads without a public HTTP endpoint. + +- **SocketMode**: WebSocket connection via `slack_sdk.socket_mode.SocketModeClient` — no Bolt framework +- **Button routing** (`slack_buttons.py`): `acknowledge`, `dispute`, `show_breakdown`, `log_decision` +- **ACE loop**: Button clicks → `_send_feedback_signal()` → `score_from_button()` (Reflector) + `update_strategy_confidence()` (Curator) +- **Decision modal**: `open_decision_modal()` captures structured decision data (decision, alternatives, reasoning) +- Location: `apps/ai/src/integrations/slack_client.py`, `apps/ai/src/integrations/slack_buttons.py` ### Goroutine Safety @@ -143,10 +192,21 @@ | `/apps/core/internal/web/templates/command_center.html` | Main dashboard template (HTMX + SSE + Chart.js) | | `/apps/core/internal/web/templates/partials/command_chat.html` | Chat panel with SSE extension | | `/apps/core/internal/web/templates/partials/command_approvals.html` | Approval queue UI (approve/hold) | -| `/apps/core/internal/web/command_center_test.go` | Test suite (52 tests covering chat, approvals, mission state, SSE) | +| `/apps/core/internal/web/command_center_test.go` | Test suite (19+ command center tests, part of 74+ web tests) | +| `/apps/core/internal/web/sse_hub.go` | SSEHub fan-out hub with event-type filtering (Subscribe/Broadcast) | | `/apps/ai/src/workflows/finance_workflow.py` | Finance specialist Temporal workflow | | `/apps/ai/src/workflows/data_workflow.py` | Data specialist Temporal workflow | | `/apps/ai/src/workflows/ops_workflow.py` | Ops specialist Temporal workflow | +| `/apps/ai/src/agents/tools/__init__.py` | ToolRegistry with ToolDef dataclass, auto-registration, tier queries | +| `/apps/ai/src/agents/tools/pause_payment_retry.py` | Tool: Pause Stripe retry (tier: review) | +| `/apps/ai/src/agents/tools/draft_investor_update.py` | Tool: Draft investor email (tier: approve) | +| `/apps/ai/src/agents/tools/schedule_customer_checkin.py` | Tool: Schedule at-risk checkin (tier: auto) | +| `/apps/ai/src/agents/tools/flag_churn_risk.py` | Tool: Flag churn risk segment (tier: auto) | +| `/apps/ai/src/integrations/slack_client.py` | Slack WebClient + SocketModeClient for interactive messages | +| `/apps/ai/src/integrations/slack_buttons.py` | ACE button routing (acknowledge, dispute, breakdown, log_decision) | +| `/apps/ai/src/hitl/manager.py` | HITLManager with route_extended for guardrail-aware tier routing | +| `/apps/ai/src/hitl/confidence.py` | Confidence scoring (pattern_seen_before, data_quality, volatility) | +| `/apps/ai/src/hitl/approval_queue.py` | Approval request/response with Slack notification | ## Architecture Evolution (2026-06-28) diff --git a/.opencode/context/domain/chat-flow.md b/.opencode/context/domain/chat-flow.md index e4b72f5..3325e65 100644 --- a/.opencode/context/domain/chat-flow.md +++ b/.opencode/context/domain/chat-flow.md @@ -137,8 +137,14 @@ func (h *Handler) APICommandChatEvents(c *fiber.Ctx) error { Key patterns: - **Fiber v2 `SetBodyStreamWriter`**: Proper SSE streaming support -- **`tryBroadcast()`**: Non-blocking send with `select/default` to prevent goroutine leaks -- **Two SSE endpoints**: `APICommandChatEvents` (chat-specific) and `APICommandEvents` (dashboard heartbeats) +- **`tryBroadcast()`**: Non-blocking send with `select/default` to prevent goroutine leaks. Also pushes to `sseHub.Broadcast("default", SSEEvent{Type: "chat", ...})` for fan-out. +- **SSEHub event-type filtering**: The SSEHub (sse_hub.go) manages per-subscriber channels with optional `Types` filter. Subscribing with `sseHub.Subscribe(tenantID, "chat")` only receives events with `Type == "chat"`. +- **Three SSE event domains** (2026-06-28): + - `event: chat` — Chat bubble HTML fragments (for `sse-swap="chat"`) + - `event: mission` — Mission state updates (prepared_brief, pending_decisions changes) + - `event: hitl` — HITL approval signals (new pending, approved, rejected) + - `event: session` — Session events (connection state, context updates) +- **SSE endpoints**: `APICommandChatEvents` (chat-specific via SSEHub), `APICommandEvents` (dashboard heartbeats), plus new typed subscribe endpoints for mission/hitl/session event types. - **`recover()`**: Deferred in stream writer to prevent panic from closed connections ### 5. HTMX SSE Swap @@ -173,12 +179,16 @@ Agent color classes: `agent-sarthi` (blue), `agent-finance` (green), `agent-data | Scenario | Behavior | File Reference | |----------|----------|----------------| -| No DB | Returns empty string | handler.go:1222-1224 | -| No Temporal | Skips workflow dispatch | handler.go:1272 | -| Workflow error | Broadcasts error bubble | handler.go:1309 | -| Channel full | Drops message, logs warning | handler.go:1415-1419 | -| Empty message | Returns empty string | handler.go:1201-1203 | -| Missing mention | No workflow dispatch, just saves message | handler.go:1259-1270 | +| No DB | Returns empty string | handler.go | +| No Temporal | Skips workflow dispatch | handler.go | +| Workflow error | Broadcasts error bubble | handler.go | +| Channel full | Drops message, logs warning | handler.go | +| Empty message | Returns empty string | handler.go | +| Missing mention | No workflow dispatch, just saves message | handler.go | +| SSEHub Subscribe with type filter | Only receives matching event types | sse_hub.go:39-53 | +| SSEHub Broadcast no matching subs | Event silently dropped (no subscriber with matching type filter) | sse_hub.go:75-93 | +| SSEHub subscriber channel full | Event dropped per sub (non-blocking per-subscriber channel) | sse_hub.go:89-92 | +| SSEHub Unsubscribe | Channel closed and removed from subscriber map | sse_hub.go:56-64 | ### 8. JavaScript Integration diff --git a/.opencode/context/standards/coding-standards.md b/.opencode/context/standards/coding-standards.md index d86ed7e..7962f60 100644 --- a/.opencode/context/standards/coding-standards.md +++ b/.opencode/context/standards/coding-standards.md @@ -66,6 +66,44 @@ func Render(c *fiber.Ctx, name string, data interface{}) error { - **Workflow result**: `run.Get(ctx, &result)` where result is `map[string]interface{}` - **Task queue**: `"TRACKGUARD-MAIN-QUEUE"` +### SSEHub Event Subscription Pattern + +The SSEHub replaces raw `chatBroadcast` channels for new SSE endpoints. Pattern: + +```go +// Subscribe with event type filter (e.g., only "chat" events) +sub := h.sseHub.Subscribe(tenantID, "chat") + +// Or subscribe to multiple event types +sub := h.sseHub.Subscribe(tenantID, "mission", "hitl") + +// Or subscribe to all events (no filter) +sub := h.sseHub.Subscribe(tenantID) +sub := h.sseHub.Subscribe(tenantID) // empty eventTypes = all events + +// Always unsubscribe in deferred cleanup +defer h.sseHub.Unsubscribe(tenantID, sub.ID) + +// Read from subscriber's channel in SetBodyStreamWriter +c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { + for { + select { + case msgBytes, ok := <-sub.Channel: + if !ok { return } + fmt.Fprintf(w, "%s", msgBytes) + w.Flush() + case <-done: + return + } + } +}) +``` + +- `SSEEvent.Type` determines routing — subscribers with matching `Types` filter receive the event +- Empty `Types` = receive all events for tenant +- Per-subscriber channel buffer: 64 (vs legacy `chatBroadcast` shared buffer of 100) +- `Broadcast()` uses non-blocking `select/default` per subscriber — a slow consumer drops their own events without affecting others + ### Concurrency Safety - `sync.WaitGroup` for goroutine tracking (workflow dispatch, graceful shutdown) @@ -137,6 +175,80 @@ with workflow.unsafe.imports_passed_through(): | Logging | `structlog` | `print()` | | Debugging | `ic` (icecream) | `print()` | +### Tool Creation Pattern (ToolDef) + +Each tool is a standalone module in `apps/ai/src/agents/tools/` with a `tool_def` dict and an async `execute` function: + +```python +"""Tool: — HITL Tier: . + +Description of what the tool does and when it triggers. +""" +from __future__ import annotations + +import logging +from typing import Any + +log = logging.getLogger(__name__) + +tool_def: dict[str, Any] = { + "name": "my_tool_name", # snake_case unique name + "description": "Human-readable description", + "hitl_tier": "review", # auto | review | approve | blocked + "trigger_patterns": ["FG-05", "BG-04"], # alert pattern IDs +} + +async def execute(tenant_id: str, **kwargs) -> dict[str, Any]: + """Execute tool logic. + + Args: + tenant_id: Tenant identifier. + **kwargs: Tool-specific parameters. + + Returns: + Dict with tool result fields. + """ + log.info("my_tool_name %s — tier=%s", tenant_id, tool_def["hitl_tier"]) + # Tool implementation + return {"status": "done", "tenant_id": tenant_id} +``` + +After creating the module, register it in `__init__.py`: + +```python +from . import my_tool_module + +register_tool(ToolDef(**my_tool_module.tool_def, fn=my_tool_module.execute)) +``` + +- `hitl_tier` must match one of the strings from `HITLManager.route()`: `"auto"`, `"review"`, `"approve"`, `"blocked"` +- `trigger_patterns` reference alert pattern IDs (FG-xx, BG-xx, etc.) +- Tools auto-register on import via `register_tool()` + +### ACE Loop Wiring Pattern + +Button interactions in Slack route through the ACE (Acknowledge-Consequence-Evaluate) loop via `slack_buttons.py`: + +```python +# In button handler: +def _handle_acknowledged(alert_id: str) -> ButtonResult: + _send_feedback_signal(alert_id, "acknowledged", 1.0) # +1.0 trust signal + return ButtonResult(success=True, action="acknowledge", ...) + +# Signal wiring (skip in test): +def _send_feedback_signal(alert_id, response_type, score): + if "pytest" in sys.modules: # skip during unit tests + return + from src.agents.cofounder.reflector import score_from_button + score_from_button(alert_id, response_type, score) + from src.agents.cofounder.curator import update_strategy_confidence + update_strategy_confidence(tenant_id, domain, response_type, score) +``` + +- Button actions: `acknowledge` (+1.0), `dispute` (-1.0), `show_breakdown`, `log_decision` +- ACE loop: Button click → Reflector scores → Curator updates strategy confidence +- Test guard: `if "pytest" in sys.modules` prevents async blocking in test envs + ### Temporal Workflow Pattern ```python diff --git a/.opencode/context/templates/architecture-overview.md b/.opencode/context/templates/architecture-overview.md index 2c7a1fb..82c0ec9 100644 --- a/.opencode/context/templates/architecture-overview.md +++ b/.opencode/context/templates/architecture-overview.md @@ -36,11 +36,15 @@ The knowledge graph identified 708 communities. Key clusters: | **Startup Guardian** | `src/guardian/` | System health monitoring, anomaly detection | | **Memory Integration** | `src/memory/` | Qdrant vector store, Redis session cache | | **LLM Ops** | `src/config/` | API key management, model selection, rate limiting | +| **Tool Registry** | `src/agents/tools/` | ToolDef dataclass, TOOL_REGISTRY, 4 registered tools with HITL tier mapping | +| **HITL Manager** | `src/hitl/` | 3-tier routing (auto/review/approve), guardrail-aware route_extended, confidence scoring | +| **Slack Integration** | `src/integrations/slack_client.py`, `slack_buttons.py` | SocketMode WebSocket client, ACE button loop, decision modal | ### Go Core Layer | Community | Key Files | Description | |-----------|-----------|-------------| -| **Go Handlers** | `handler.go`, `sse.go` | HTTP endpoints, SSE streaming, HTMX partials | +| **Go Handlers** | `handler.go`, `sse.go`, `sse_hub.go` | HTTP endpoints, SSE streaming (legacy polling), SSEHub event-type fan-out | +| **SSEHub** | `sse_hub.go` | Per-subscriber channel hub with event-type filtering (`Subscribe`/`Broadcast`) | | **Temporal Workflows** | `internal/workflow/`, `internal/temporal/` | Workflow stubs, client, activity definitions | | **Go Database Layer** | `internal/db/`, `internal/database/` | SQL queries, connection management | | **Config Module** | `internal/config/` | LLM provider config (Azure, Groq, Ollama) | @@ -110,8 +114,11 @@ make build # Builds Go binaries |-------|---------|-------------| | `GET /command` | `CommandCenter` | Main dashboard page | | `POST /api/command/chat/send` | `APICommandChatSend` | Chat submission + Temporal dispatch in goroutine | -| `GET /api/command/chat/events` | `APICommandChatEvents` | SSE stream for chat bubbles (SetBodyStreamWriter) | -| `GET /api/command/events` | `APICommandEvents` | SSE stream for dashboard heartbeats | +| `GET /api/command/chat/events` | `APICommandChatEvents` | SSE stream for chat bubbles (SSEHub, subscribes to `chat` events) | +| `GET /api/command/events` | `APICommandEvents` | SSE stream for dashboard heartbeats (legacy, no SSEHub) | +| `GET /api/command/mission/events` | `APICommandMissionEvents` | SSE stream for mission state updates (SSEHub, subscribes to `mission` events) | +| `GET /api/command/hitl/events` | `APICommandHITLEvents` | SSE stream for HITL approval signals (SSEHub, subscribes to `hitl` events) | +| `GET /api/command/session/events` | `APICommandSessionEvents` | SSE stream for session context events (SSEHub, subscribes to `session` events) | | `GET /api/command/status` | `APICommandStatus` | Health score bar | | `GET /api/command/kpis` | `APICommandKPIs` | KPI cards (MRR, Runway, etc.) | | `GET /api/command/mission-state` | `APICommandMissionState` | Read mission_state for dashboard | @@ -128,8 +135,8 @@ make build # Builds Go binaries ```sql -- Core operational tables (command_center.sql) -mission_state -- Compiled state from Python AI (MRR, burn, health) -planned_actions -- Approval queue (actor, action_type, risk_level) +mission_states -- Compiled state from Python AI (MRR, burn, health, brief, decisions) +planned_actions -- Approval queue (actor, action_type, risk_level, workflow_id) agent_traces -- Activity log from Python AI (duration, tokens, cost) chat_messages -- Chat history (sender, mention, message) @@ -139,6 +146,8 @@ agent_outputs -- Finance anomaly alerts, BI query results agent_events -- Polled SSE events for legacy dashboard ``` +**Note:** `mission_state` → `mission_states` (migration 004 resolved schema drift). The MissionState dataclass (`apps/ai/src/session/mission_state.py`) now includes `prepared_brief`, `pending_decisions`, and `last_updated_by` fields. + ## Quick Reference ### Adding a New Specialist Workflow diff --git a/AGENTS.md b/AGENTS.md index 0a43169..3c7464a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,13 +18,17 @@ |-----------|--------|-------| | Python Tests | ✅ 319/320 passing | 1 pre-existing timeout in curator_graphiti | | Go Build | ✅ Clean | Binary compiles successfully | -| Go HTMX Handler Tests | ✅ 52 passing | command center, chat, approvals, mission state, SSE | +| Go HTMX Handler Tests | ✅ 74+ passing | command center (19), business handlers (13), founder (14), plus LLMops, onboarding, watchlist, telegram, razorpay | | HTMX Routes | ✅ 13+ routes | Command center dashboard panels + SSE endpoints | -| SSE Streaming | ✅ | HTMX `hx-ext="sse"` + `SetBodyStreamWriter` pattern | +| SSE Streaming | ✅ | HTMX `hx-ext="sse"` + `SetBodyStreamWriter` pattern + SSEHub event-type filtering | +| SSEHub Event-Type Filtering | ✅ | `Subscribe(tenantID, eventTypes...)` with per-subscriber channels (buffered 64) | | Specialist Agents | ✅ | Finance, Data, Ops (each with LangGraph graph + Temporal workflow) | | HITL Temporal Signals | ✅ | `SignalWorkflow("hitl-approval")` unblocks `AwaitWithTimeout` | | MissionState Write Path | ✅ | `POST /api/mission-state` from Python AI → PostgreSQL | | @mention Routing | ✅ | `map[string]specialistRoute` — O(1) map lookup, 9 aliases → 6 workflows | +| Tool Calling Surface | ✅ | `apps/ai/src/agents/tools/` — ToolDef, TOOL_REGISTRY, 4 tools with HITL tier mapping | +| ACE Reflector Loop | ✅ | Slack button clicks → `score_from_button()` + `update_strategy_confidence()` | +| MissionState Promotion | ✅ | Added `prepared_brief`, `pending_decisions`, `last_updated_by` fields | | DB Tests | 🟡 Skip | Requires PostgreSQL container | | Webhook Tests | 🟡 Skip | Requires Redpanda container | @@ -149,6 +153,7 @@ uv run pytest tests/ --cov=src --cov-report=term-missing - HTMX declarative: `hx-ext="sse"` + `sse-connect` + `sse-swap` + `hx-swap` - HTML fragments as SSE data payload (server-rendered via `renderChatBubble()`) - Always `html.EscapeString()` on user/LLM text for XSS protection +- SSEHub event-type filtering: `Subscribe(tenantID, eventTypes...)` + `Broadcast(tenantID, SSEEvent)` **Goroutine Safety:** - `sync.WaitGroup` for tracking in-flight workflow dispatches @@ -160,13 +165,19 @@ uv run pytest tests/ --cov=src --cov-report=term-missing ```go var specialistRoutes = map[string]specialistRoute{ "@sarthi": {"QAWorkflow", "Sarthi"}, + "@agent": {"QAWorkflow", "Sarthi"}, + "@qa": {"QAWorkflow", "Sarthi"}, + "@ask": {"QAWorkflow", "Sarthi"}, "@finance": {"FinanceWorkflow", "Finance"}, "@data": {"DataWorkflow", "Data"}, "@ops": {"OpsWorkflow", "Ops"}, + "@comms": {"CommsWorkflow", "Comms"}, + "@hiring": {"HiringWorkflow", "Hiring"}, } ``` - O(1) map lookup replaces if-else chain - Adding a specialist = 1 map entry + 1 Python workflow class +- 9 aliases → 6 workflows (QA, Finance, Data, Ops, Comms, Hiring) ### SQL (sqlc) @@ -228,6 +239,12 @@ apps/ finance/ # V4 NEW — Finance specialist (FinanceGraph) data/ # V4 NEW — Data specialist (DataGraph) ops/ # V4 NEW — Ops specialist (OpsGraph) + tools/ # V4 NEW — ToolRegistry + 4 ToolDef implementations + __init__.py # ToolDef, TOOL_REGISTRY, register_tool(), auto-import + pause_payment_retry.py # FG-05: pause Stripe retry (review) + draft_investor_update.py # Schedule: draft investor email (approve) + schedule_customer_checkin.py # FG-03/BG-04: at-risk checkin (auto) + flag_churn_risk.py # BG-06/BG-04: flag churn segment (auto) workflows/ # V4 NEW — Temporal workflow definitions finance_workflow.py data_workflow.py diff --git a/README.md b/README.md index a31f003..0af3b73 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,10 @@ -# Sarthi — AI Coordination Layer for Solo Founders +# Sarthi — Handles everything up to the moment of consequential judgment > Server-rendered command center with SSE push, goroutine-based Temporal dispatch, and Python specialist agents. > Chat → @mention → specialist workflow → SSE result — all driven by Go + Temporal + LangGraph. +> Sarthi handles everything up to the moment of consequential judgment — the AI coordination layer for solo founders. -[![Tests](https://img.shields.io/badge/tests-371%20passing-brightgreen)](#) +[![Tests](https://img.shields.io/badge/tests-393%2B%20passing-brightgreen)](#) [![Architecture](https://img.shields.io/badge/architecture-SSE%20%2B%20Specialist-blue)](#) [![Go](https://img.shields.io/badge/Go-1.24-blue?logo=go)](#) [![Python](https://img.shields.io/badge/Python-3.13-green?logo=python)](#) @@ -121,6 +122,9 @@ User types "@finance Q3 revenue?" → HTMX POST /api/command/chat/send | **Server-rendered chat bubbles** | `html.EscapeString()` XSS protection. Agent color classes. Single source of truth for HTML. | | **MissionState POST endpoint** | Python AI → POST → PostgreSQL → GET → Dashboard. Pure server-side rendered. | | **Remove dead stubs** | Cleaned 40 lines of stale placeholder types from `workflow/stubs.go`. | +| **SSEHub event-type filtering** | Event-type routing (chat/mission/hitl/session). Per-subscriber channels (64 buffer) eliminate head-of-line blocking. | +| **ToolRegistry + HITL tier mapping** | 4 tools with explicit tiers (auto/review/approve) triggered by alert pattern IDs. Self-documenting tool definitions. | +| **Slack SocketMode + ACE loop** | No Bolt dependency. Socket Mode eliminates public HTTP endpoint. Button clicks → Reflector → Curator confidence. | > Full details: [ADR-001: Sarthi v4.0 Architecture Evolution](.opencode/context/adr/001-sarthi-v4-architecture-evolution.md) @@ -158,6 +162,7 @@ var specialistRoutes = map[string]specialistRoute{ "@hiring": {"HiringWorkflow", "Hiring"}, } ``` +- O(1) map lookup, 9 aliases → 6 workflows. --- @@ -178,11 +183,31 @@ var specialistRoutes = map[string]specialistRoute{ - User clicks Approve → POST → `SignalWorkflow(ctx, id, "hitl-approval", true)` - Workflow unblocks, execution continues +### Tool Calling Surface (ToolRegistry) +4 tool functions defined as `ToolDef` entries in a global `TOOL_REGISTRY`, wired to HITL manager: + +| Tool | Tier | Trigger | Action | +|------|------|---------|--------| +| `pause_failed_payment_retry` | review | FG-05 (3+ failed payments) | Pause Stripe retry | +| `draft_investor_update` | approve | Schedule | Draft investor email | +| `schedule_customer_checkin` | auto | FG-03, BG-04 | Auto-schedule reminder | +| `flag_churn_risk_customer` | auto | BG-06, BG-04 | Flag churn segment | + +Tools auto-register via `register_tool(ToolDef(...))` on import. `get_tools_for_tier()` and `get_tools_for_patterns()` enable pattern-driven tool suggestion. + +### ACE Reflector Loop (Slack Integration) +- `SlackClient` extended with `SocketModeClient` (WebSocket, no Bolt) +- Button interactions routed in `slack_buttons.py` (5 action types) +- ACE loop: button click → `score_from_button()` (Reflector) → `update_strategy_confidence()` (Curator) +- Decision modal captures structured decisions (decision, alternatives, reasoning) + ### MissionState Write Path - **Python AI** compiles operational state (MRR, burn, health, signals) -- **POST** to `/api/mission-state` → **PostgreSQL** (`mission_state` table) +- **POST** to `/api/mission-state` → **PostgreSQL** (`mission_states` table) - **GET** → Go templates → HTML (dashboard) - Pure server-side rendered — no client state +- New fields: `prepared_brief`, `pending_decisions`, `last_updated_by` +- Schema: `mission_states` (migration 004 reconciled `mission_state` → `mission_states`) ### SSE Chat System - HTMX `hx-ext="sse"` declaratively subscribes to SSE stream @@ -190,7 +215,10 @@ var specialistRoutes = map[string]specialistRoute{ - `renderChatBubble()` with `html.EscapeString()` — XSS-safe - Agent color classes: `agent-sarthi` (blue), `agent-finance` (green), `agent-data` (purple), `agent-ops` (yellow) - Non-blocking `tryBroadcast()` with `select/default` on buffered channel (capacity 100) -- Two SSE endpoints: chat-specific and dashboard heartbeat +- **SSEHub** (`sse_hub.go`): Event-type filtered fan-out hub with per-subscriber channels (buffer 64) + - `Subscribe(tenantID, eventTypes...)` — typed subscriptions (chat, mission, hitl, session) + - `Broadcast(tenantID, SSEEvent)` — delivers only to matching subscribers +- Event types: `chat` (bubbles), `mission` (state updates), `hitl` (approval signals), `session` (context events) ### Goroutine Safety Patterns - `sync.WaitGroup` for graceful shutdown tracking of in-flight workflow dispatches @@ -220,12 +248,12 @@ The V3.0 deterministic business logic layer (Finance Rules, Guardrails Engine, P --- -## Test Coverage (371+ Passing — Go Build Clean) +## Test Coverage (393+ Passing — Go Build Clean) | Suite | Tests | Status | |-------|-------|--------| | Python Unit Tests | 319 | ✅ (1 pre-existing timeout in curator_graphiti skipped) | -| Go HTMX Web Handlers | 52 | ✅ | +| Go HTMX Web Handlers | 74+ | ✅ | | Go Build | Clean | ✅ | | DB Tests | 🟡 Skip | Requires PostgreSQL container | | Redpanda Tests | 🟡 Skip | Requires Redpanda container | @@ -239,12 +267,13 @@ The V3.0 deterministic business logic layer (Finance Rules, Guardrails Engine, P - Deterministic Trajectory (87), State Machine (17), Edge Cases (47), Contracts (68), Mockoon (17) - All Others (100+) -**Go Web Handler Suites (52 tests):** -- Command Center (chat, approvals, mission state) -- SSE streaming endpoints -- @mention routing and specialist dispatch -- HITL approval signal flow -- Template rendering +**Go Web Handler Suites (74+ tests):** +- Command Center (chat, approvals, mission state) — 19 tests +- Business Handlers (decision queue, guardrail status, finance risk) — 13 tests +- Founder Dashboard (summary, reflection, commitments) — 14 tests +- LLM Ops Dashboard, Onboarding Status, Watchlist Viewer — 9 tests +- Telegram, Razorpay webhook handlers — 18 tests +- SSE streaming and chat broadcast — 1 test --- @@ -287,6 +316,7 @@ apps/ finance/ # Finance specialist (V4 NEW — FinanceGraph) data/ # Data specialist (V4 NEW — DataGraph) ops/ # Ops specialist (V4 NEW — OpsGraph) + tools/ # V4 NEW — ToolRegistry + 4 ToolDef implementations business/ # V3.0 MBA integration (Finance Rules, Guardrails) predictive/ # V3.0 Forecasting engine workflows/ # V4 NEW — Temporal workflow definitions diff --git a/apps/ai/infrastructure/migrations/004_control_plane_audit.sql b/apps/ai/infrastructure/migrations/004_control_plane_audit.sql new file mode 100644 index 0000000..3e5e42f --- /dev/null +++ b/apps/ai/infrastructure/migrations/004_control_plane_audit.sql @@ -0,0 +1,30 @@ +-- Migration 004: Add control plane audit logging +-- Creates the audit_log table for tracking all control-plane-gated actions +CREATE TABLE IF NOT EXISTS audit_log ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + tenant_id VARCHAR(100) NOT NULL, + agent_name VARCHAR(200) NOT NULL, + action VARCHAR(100) NOT NULL, + tool_name VARCHAR(200), + model_used VARCHAR(100), + policy_decision JSONB, + approval_state VARCHAR(20), + outcome VARCHAR(20) NOT NULL, + details JSONB, + timestamp TIMESTAMPTZ DEFAULT NOW(), + created_at TIMESTAMPTZ DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_audit_log_tenant_time + ON audit_log(tenant_id, timestamp DESC); + +CREATE INDEX IF NOT EXISTS idx_audit_log_agent + ON audit_log(agent_name, timestamp DESC); + +CREATE INDEX IF NOT EXISTS idx_audit_log_outcome + ON audit_log(outcome, timestamp DESC); + +COMMENT ON TABLE audit_log IS 'Control plane audit trail for all agent actions gated by policy, risk, and HITL routing'; +COMMENT ON COLUMN audit_log.policy_decision IS 'Snapshot of the PolicyDecision that governed this action'; +COMMENT ON COLUMN audit_log.approval_state IS 'auto | review | approve | blocked'; +COMMENT ON COLUMN audit_log.outcome IS 'completed | blocked | failed'; diff --git a/apps/ai/pyproject.toml b/apps/ai/pyproject.toml index a4f519c..a9db37b 100644 --- a/apps/ai/pyproject.toml +++ b/apps/ai/pyproject.toml @@ -56,6 +56,8 @@ dependencies = [ "ollama>=0.6.2", "google-genai>=2.2.0", "hubspot-api-client", + # Slack SDK (SocketModeClient + WebClient for interactive events & alerts) + "slack-sdk>=3.30.0", ] [project.optional-dependencies] diff --git a/apps/ai/src/agents/authority_manifest.py b/apps/ai/src/agents/authority_manifest.py new file mode 100644 index 0000000..d6dfb46 --- /dev/null +++ b/apps/ai/src/agents/authority_manifest.py @@ -0,0 +1,124 @@ +"""Agent Authority Manifest — declarative capability/escalation registry. + +Defines each agent's domain, tool permissions, escalation tier, triggers, +and MissionState fields it is allowed to write. Used by HITL routing, +tool execution guards, and MissionState write-path validation. +""" +from pydantic import BaseModel +from typing import Literal + + +class AgentAuthority(BaseModel): + agent_name: str + role: str + voice: str + domain: Literal["finance", "bi", "ops", "cofounder", "correlation"] + can_emit_alerts: bool + can_execute_tools: bool + allowed_tool_ids: list[str] + escalation_tier: Literal["auto", "review", "approve", "blocked"] + triggers: list[str] + writes_mission_fields: list[str] + allowed_models: list[str] = ["gpt-4o-mini"] + external_facing: bool = False + data_classification: str = "internal" + + +AUTHORITY_MANIFEST: list[AgentAuthority] = [ + AgentAuthority( + agent_name="Sarthi", + role="manager/cofounder", + voice="Founder's strategic thinking partner", + domain="cofounder", + can_emit_alerts=True, + can_execute_tools=True, + allowed_tool_ids=["draft_investor_update"], + escalation_tier="approve", + triggers=["manual", "schedule"], + writes_mission_fields=["founder_focus", "active_alerts", "prepared_brief"], + allowed_models=["gpt-4o"], + external_facing=True, + data_classification="external_investor", + ), + AgentAuthority( + agent_name="Sarthi · Finance", + role="Finance specialist", + voice="Financial guardian", + domain="finance", + can_emit_alerts=True, + can_execute_tools=True, + allowed_tool_ids=["pause_failed_payment_retry"], + escalation_tier="review", + triggers=["FG-01", "FG-02", "FG-03", "FG-04", "FG-05", "FG-06"], + writes_mission_fields=["runway_days", "burn_alert", "burn_severity", "burn_multiple"], + allowed_models=["gpt-4o-mini"], + external_facing=False, + data_classification="internal", + ), + AgentAuthority( + agent_name="Sarthi · Data", + role="BI specialist", + voice="Data analyst", + domain="bi", + can_emit_alerts=True, + can_execute_tools=True, + allowed_tool_ids=["draft_investor_update"], + escalation_tier="auto", + triggers=["BG-01", "BG-02", "BG-03", "BG-04", "BG-05", "BG-06"], + writes_mission_fields=["mrr_trend", "churn_rate"], + allowed_models=["gpt-4o-mini"], + external_facing=False, + data_classification="internal", + ), + AgentAuthority( + agent_name="Sarthi · Ops", + role="Ops specialist", + voice="Operations watchdog", + domain="ops", + can_emit_alerts=True, + can_execute_tools=True, + allowed_tool_ids=["flag_churn_risk_customer", "schedule_customer_checkin"], + escalation_tier="auto", + triggers=["OG-01", "OG-02", "OG-03", "OG-04", "OG-05"], + writes_mission_fields=["churn_risk_users", "top_feature_ask", "error_spike"], + allowed_models=["gpt-4o-mini"], + external_facing=False, + data_classification="internal", + ), + AgentAuthority( + agent_name="Correlation Agent", + role="Cross-domain synthesizer", + voice="Pattern detector", + domain="correlation", + can_emit_alerts=True, + can_execute_tools=False, + allowed_tool_ids=[], + escalation_tier="review", + triggers=["cross-domain"], + writes_mission_fields=["active_alerts"], + allowed_models=["gpt-4o-mini"], + external_facing=False, + data_classification="internal", + ), +] + + +def get_authority(agent_name: str) -> AgentAuthority | None: + for a in AUTHORITY_MANIFEST: + if a.agent_name == agent_name: + return a + return None + + +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 + + +def get_writes_mission_fields(agent_name: str) -> list[str]: + auth = get_authority(agent_name) + if auth is None: + return [] + return auth.writes_mission_fields diff --git a/apps/ai/src/agents/cofounder/curator.py b/apps/ai/src/agents/cofounder/curator.py index af13692..f5308e0 100644 --- a/apps/ai/src/agents/cofounder/curator.py +++ b/apps/ai/src/agents/cofounder/curator.py @@ -16,6 +16,8 @@ from datetime import datetime, timezone from typing import Optional +from pydantic import BaseModel + from src.memory.semantic import SemanticMemory log = logging.getLogger(__name__) @@ -286,13 +288,37 @@ class ConfidenceUpdateResult: error: str | None = None +class StrategyDelta(BaseModel): + """Immutable audit record produced by every update_strategy_confidence call. + + Written to /tmp/strategy_audit.jsonl as a fallback when PostgreSQL is + unavailable. The linked_alert_id is the strategy_key for lineage tracing. + """ + tenant_id: str + domain: str + strategy_key: str + prior_confidence: float + new_confidence: float + evidence_count_delta: int + feedback_type: str + linked_alert_id: str + + def update_strategy_confidence( tenant_id: str, domain: str, feedback_type: str, score: float, -) -> ConfidenceUpdateResult: - """Update Graphiti Strategy confidence score based on founder feedback.""" + strategy_key: str = "", + linked_alert_id: str = "", + prior_confidence: float = 0.0, +) -> tuple[ConfidenceUpdateResult, StrategyDelta]: + """Update Graphiti Strategy confidence score based on founder feedback. + + Returns: + Tuple of (ConfidenceUpdateResult, StrategyDelta). The StrategyDelta + is always produced — even on failure — for audit lineage. + """ score_map = { "acknowledged": 1.0, "acted_on": 1.5, @@ -301,6 +327,18 @@ def update_strategy_confidence( "dismissed": -1.5, } delta = score_map.get(feedback_type, score) + new_confidence = prior_confidence + delta + + delta_record = StrategyDelta( + tenant_id=tenant_id, + domain=domain, + strategy_key=strategy_key, + prior_confidence=prior_confidence, + new_confidence=new_confidence, + evidence_count_delta=1, + feedback_type=feedback_type, + linked_alert_id=linked_alert_id or strategy_key, + ) try: from src.memory.semantic import SemanticMemory @@ -317,13 +355,17 @@ def update_strategy_confidence( name=f"strategy_confidence:{domain}", body=playbook_entry, ) + + # Audit log: try PostgreSQL first, fall back to file append + _write_audit_log(delta_record) + return ConfidenceUpdateResult( success=True, tenant_id=tenant_id, domain=domain, confidence_delta=delta, - new_confidence=delta, - ) + new_confidence=new_confidence, + ), delta_record except Exception as e: return ConfidenceUpdateResult( success=False, @@ -331,4 +373,72 @@ def update_strategy_confidence( domain=domain, confidence_delta=delta, error=str(e), - ) \ No newline at end of file + ), delta_record + + +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) + + +def _file_audit_write(delta: StrategyDelta) -> None: + """Fallback: append StrategyDelta as JSON line to /tmp/strategy_audit.jsonl.""" + import os + + audit_path = "/tmp/strategy_audit.jsonl" + line = delta.model_dump_json() + "\n" + with open(audit_path, "a") as f: + f.write(line) + log.info(f"[Audit] Wrote strategy delta to {audit_path}: {delta.domain}/{delta.strategy_key}") \ No newline at end of file diff --git a/apps/ai/src/agents/tools/__init__.py b/apps/ai/src/agents/tools/__init__.py new file mode 100644 index 0000000..d0805c8 --- /dev/null +++ b/apps/ai/src/agents/tools/__init__.py @@ -0,0 +1,68 @@ +"""Tool registry for Sarthi agent actions. + +Each tool is a standalone async function with a tool_def dict for metadata. +Tools are registered in TOOL_REGISTRY and wired to HITL manager for resolution. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Coroutine + + +@dataclass +class ToolDef: + """Definition of a callable tool with HITL tier routing metadata. + + Attributes: + name: Unique tool name (snake_case). + description: Human-readable description of what the tool does. + hitl_tier: HITL tier — one of "auto", "review", "approve", "blocked". + Must match the string returned by HITLManager.route(). + fn: Async function that executes the tool. Takes tenant_id as first arg. + trigger_patterns: Alert pattern IDs that suggest this tool + (e.g. "FG-05", "BG-04"). + """ + name: str + description: str + hitl_tier: str + fn: Callable[..., Coroutine[Any, Any, dict[str, Any]]] + trigger_patterns: list[str] = field(default_factory=list) + + +TOOL_REGISTRY: dict[str, ToolDef] = {} + + +def register_tool(tool: ToolDef) -> None: + """Register a tool in the global TOOL_REGISTRY.""" + TOOL_REGISTRY[tool.name] = tool + + +def get_tools_for_tier(tier: str) -> list[ToolDef]: + """Get all tools whose HITL tier matches the given routing decision.""" + return [t for t in TOOL_REGISTRY.values() if t.hitl_tier == tier] + + +def get_tools_for_pattern(pattern_id: str) -> list[ToolDef]: + """Get all tools whose trigger_patterns include the given pattern ID.""" + return [t for t in TOOL_REGISTRY.values() if pattern_id in t.trigger_patterns] + + +def get_tools_for_patterns(pattern_ids: list[str]) -> list[ToolDef]: + """Get all tools matching any of the given triggered pattern IDs.""" + matched: dict[str, ToolDef] = {} + for pid in pattern_ids: + for t in TOOL_REGISTRY.values(): + if pid in t.trigger_patterns and t.name not in matched: + matched[t.name] = t + return list(matched.values()) + + +# Auto-register all tool modules on import +from . import pause_payment_retry +from . import draft_investor_update +from . import schedule_customer_checkin +from . import flag_churn_risk + +for _mod in [pause_payment_retry, draft_investor_update, + schedule_customer_checkin, flag_churn_risk]: + register_tool(ToolDef(**_mod.tool_def, fn=_mod.execute)) diff --git a/apps/ai/src/agents/tools/draft_investor_update.py b/apps/ai/src/agents/tools/draft_investor_update.py new file mode 100644 index 0000000..2eecd7c --- /dev/null +++ b/apps/ai/src/agents/tools/draft_investor_update.py @@ -0,0 +1,127 @@ +"""Tool: draft_investor_update — HITL Tier: approve. + +Always requires founder approval before sending. Assembles the +latest mission state metrics into an investor-facing update email. + +Integrates prompt risk scanning (pre-generation), output risk scanning +(post-generation), and audit logging via the control plane. +""" +from __future__ import annotations + +import logging +from typing import Any + +from src.config.llm import chat_completion +from src.control_plane.audit import AuditLogger +from src.control_plane.policy import PolicyEngine +from src.control_plane.registry import ControlPlaneRegistry +from src.risk.prompt_risk import scan_prompt +from src.risk.output_risk import scan_output +from src.schemas.control_plane import AgentRegistration, AuditEvent +from src.session.mission_state import get_mission_state + +log = logging.getLogger(__name__) + +# Module-level control plane instances (registration happens at app startup) +_REGISTRY = ControlPlaneRegistry() +_POLICY = PolicyEngine() +_AUDIT = AuditLogger() + +tool_def: dict[str, Any] = { + "name": "draft_investor_update", + "description": "Draft an investor update email for founder approval", + "hitl_tier": "approve", + "trigger_patterns": ["schedule", "manual"], +} + + +async def execute(tenant_id: str) -> dict[str, Any]: + """Draft investor update with prompt/output risk scanning and audit logging. + + Args: + tenant_id: Tenant identifier. + + Returns: + Dict with draft text, risk scan results, audit event, and + requires_approval flag. + """ + log.info("draft_investor_update %s — tier=approve (risk-gated)", tenant_id) + state = await get_mission_state(tenant_id) + + metrics = ( + f"Runway: {state.runway_days}d | " + f"MRR trend: {state.mrr_trend} | " + f"Churn: {state.churn_rate} | " + f"Trust: {state.trust_score}" + ) + metrics_prompt = ( + f"Write a brief investor update email draft based on: {metrics}. " + "Professional tone, 2-3 paragraphs." + ) + + # ── Pre-generation prompt risk scan ────────────────────────────── + prompt_scan = scan_prompt(metrics_prompt, context="investor_update") + if prompt_scan.recommended_action == "block": + log.warning("draft_investor_update %s — prompt risk scan blocked", tenant_id) + return { + "error": "Prompt risk scan blocked", + "scan_result": prompt_scan.model_dump(), + "tenant_id": tenant_id, + "requires_approval": True, + } + + # ── LLM call ───────────────────────────────────────────────────── + draft = chat_completion( + messages=[{"role": "user", "content": metrics_prompt}], + max_tokens=300, + temperature=0.5, + ) + + # ── Post-generation output risk scan ───────────────────────────── + output_scan = scan_output(draft, context="investor_update") + if output_scan.recommended_action == "block": + log.warning("draft_investor_update %s — output risk scan blocked", tenant_id) + audit_event = AuditEvent( + agent_name="Sarthi", + action="tool_execution", + tool_name="draft_investor_update", + model_used="gpt-4o", + approval_state="blocked", + outcome="blocked", + tenant_id=tenant_id, + details={ + "output_scan": output_scan.model_dump(), + "reason": "Output risk scan blocked draft", + }, + ) + _AUDIT.log(audit_event) + return { + "draft": draft, + "risk_scan": output_scan.model_dump(), + "tenant_id": tenant_id, + "requires_approval": True, + "message": "Draft blocked by output risk scan — requires approval", + } + + # ── Audit log on successful generation ─────────────────────────── + audit_event = AuditEvent( + agent_name="Sarthi", + action="tool_execution", + tool_name="draft_investor_update", + model_used="gpt-4o", + approval_state="approve", + outcome="completed", + tenant_id=tenant_id, + details={ + "prompt_scan": prompt_scan.model_dump(), + "output_scan": output_scan.model_dump(), + }, + ) + _AUDIT.log(audit_event) + + return { + "draft": draft, + "tenant_id": tenant_id, + "requires_approval": True, + "audit_event": audit_event.model_dump(), + } diff --git a/apps/ai/src/agents/tools/flag_churn_risk.py b/apps/ai/src/agents/tools/flag_churn_risk.py new file mode 100644 index 0000000..5e242fd --- /dev/null +++ b/apps/ai/src/agents/tools/flag_churn_risk.py @@ -0,0 +1,47 @@ +"""Tool: flag_churn_risk_customer — HITL Tier: auto. + +Auto-executes for BG-06 (trial activation wall) or BG-04 (cohort +retention degradation). Flags the segment in the customer DB so +the dashboard and alerting pipeline can prioritize monitoring. +""" +from __future__ import annotations + +import logging +from typing import Any + +from src.session.mission_state import get_mission_state, update_mission_state + +log = logging.getLogger(__name__) + +tool_def: dict[str, Any] = { + "name": "flag_churn_risk_customer", + "description": "Flag a customer segment as churn risk in the database", + "hitl_tier": "auto", + "trigger_patterns": ["BG-06", "BG-04"], +} + + +async def execute(tenant_id: str, segment_id: str) -> dict[str, Any]: + """Flag a segment for churn risk monitoring. + + Args: + tenant_id: Tenant identifier. + segment_id: Customer segment ID to flag. + + Returns: + Dict with flagged status, segment_id, and tenant_id. + """ + log.info("flag_churn_risk_customer %s/%s — tier=auto", + tenant_id, segment_id) + state = await get_mission_state(tenant_id) + existing = state.churn_risk_users or "" + if segment_id not in existing: + state.churn_risk_users = (existing + "," + segment_id).strip(",") + state.last_updated_by = "flag_churn_risk_tool" + await update_mission_state(state) + log.info({"action": "flag_churn_risk", "segment_id": segment_id}) + return { + "flagged": True, + "segment_id": segment_id, + "tenant_id": tenant_id, + } diff --git a/apps/ai/src/agents/tools/pause_payment_retry.py b/apps/ai/src/agents/tools/pause_payment_retry.py new file mode 100644 index 0000000..fc8737b --- /dev/null +++ b/apps/ai/src/agents/tools/pause_payment_retry.py @@ -0,0 +1,64 @@ +"""Tool: pause_failed_payment_retry — HITL Tier: review. + +Triggered by FG-05 (3+ failed payments in 7 days). Pauses Stripe's +automatic retry schedule so the founder can investigate before +the customer is charged again. +""" +from __future__ import annotations + +import logging +import os +from typing import Any + +import httpx + +from src.integrations.stripe import MOCK_MODE + +log = logging.getLogger(__name__) + +tool_def: dict[str, Any] = { + "name": "pause_failed_payment_retry", + "description": "Pause Stripe retry schedule for a customer with failed payments", + "hitl_tier": "review", + "trigger_patterns": ["FG-05"], +} + + +async def execute(tenant_id: str, subscription_id: str) -> dict[str, Any]: + """Pause automatic retry on a failed payment. + + Args: + tenant_id: Tenant identifier. + subscription_id: Stripe subscription ID whose retries should be paused. + + Returns: + Dict with status, subscription_id, and tenant_id. + """ + log.info("pause_failed_payment_retry %s/%s — tier=review", tenant_id, subscription_id) + if MOCK_MODE: + return { + "status": "paused", + "subscription_id": subscription_id, + "tenant_id": tenant_id, + "mock": True, + } + api_key = os.getenv("STRIPE_API_KEY", "") + headers = { + "Authorization": f"Bearer {api_key}", + "Stripe-Version": "2023-10-16", + } + url = f"https://api.stripe.com/v1/subscriptions/{subscription_id}" + async with httpx.AsyncClient() as client: + resp = await client.post( + url, + data={"collection_method": "send_invoice"}, + headers=headers, + timeout=30.0, + ) + resp.raise_for_status() + return { + "status": "paused", + "subscription_id": subscription_id, + "tenant_id": tenant_id, + "response": resp.json(), + } diff --git a/apps/ai/src/agents/tools/schedule_customer_checkin.py b/apps/ai/src/agents/tools/schedule_customer_checkin.py new file mode 100644 index 0000000..86f5695 --- /dev/null +++ b/apps/ai/src/agents/tools/schedule_customer_checkin.py @@ -0,0 +1,57 @@ +"""Tool: schedule_customer_checkin — HITL Tier: auto. + +Auto-executes for FG-03 (customer concentration risk) or BG-04 +(cohort retention degradation). Logs the decision in the decision +journal and schedules a Slack reminder. +""" +from __future__ import annotations + +import logging +import os +from typing import Any + +from src.integrations.slack_client import SlackClient + +log = logging.getLogger(__name__) + +tool_def: dict[str, Any] = { + "name": "schedule_customer_checkin", + "description": "Schedule a follow-up reminder for an at-risk customer", + "hitl_tier": "auto", + "trigger_patterns": ["FG-03", "BG-04"], +} + + +async def execute(tenant_id: str, customer_id: str, days_out: int = 7) -> dict[str, Any]: + """Schedule a check-in reminder. + + Args: + tenant_id: Tenant identifier. + customer_id: Customer ID to check in with. + days_out: Days until the reminder fires (default 7). + + Returns: + Dict with scheduled status, customer_id, days_out, and tenant_id. + """ + log.info("schedule_customer_checkin %s/%s — tier=auto, days=%d", + tenant_id, customer_id, days_out) + if not os.getenv("SLACK_BOT_TOKEN"): + return { + "scheduled": True, + "customer_id": customer_id, + "days_out": days_out, + "tenant_id": tenant_id, + "mock": True, + } + client = SlackClient() + channel = os.getenv("SLACK_CHECKIN_CHANNEL", "#customer-checkins") + msg = f"*Check-in reminder:* Follow up with customer `{customer_id}` in {days_out} days." + resp = client.client.chat_postMessage(channel=channel, text=msg) + return { + "scheduled": True, + "customer_id": customer_id, + "days_out": days_out, + "tenant_id": tenant_id, + "channel": channel, + "ts": resp.get("ts", ""), + } diff --git a/apps/ai/src/control_plane/__init__.py b/apps/ai/src/control_plane/__init__.py new file mode 100644 index 0000000..dd83ed4 --- /dev/null +++ b/apps/ai/src/control_plane/__init__.py @@ -0,0 +1,10 @@ +"""Control Plane — shared agent registry, policy engine, and audit logging.""" +from src.control_plane.registry import ControlPlaneRegistry +from src.control_plane.policy import PolicyEngine +from src.control_plane.audit import AuditLogger + +__all__ = [ + "ControlPlaneRegistry", + "PolicyEngine", + "AuditLogger", +] \ No newline at end of file diff --git a/apps/ai/src/control_plane/audit.py b/apps/ai/src/control_plane/audit.py new file mode 100644 index 0000000..a0cda5f --- /dev/null +++ b/apps/ai/src/control_plane/audit.py @@ -0,0 +1,78 @@ +"""Audit Logger — writes control plane audit events to PostgreSQL.""" +from __future__ import annotations + +import json +import logging + +import asyncpg + +from src.config.database import get_database_url +from src.schemas.control_plane import AuditEvent + +log = logging.getLogger(__name__) + +DATABASE_URL = get_database_url("iterateswarm") + + +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: + """Persist an audit event to PostgreSQL. + + Args: + event: The AuditEvent to log. + conn: Optional existing connection for transaction reuse. + + Returns: + True if logged successfully, False on failure. + """ + own_conn = conn is None + try: + if own_conn: + conn = await asyncpg.connect(DATABASE_URL) + + await conn.execute( + """ + INSERT INTO audit_log ( + tenant_id, agent_name, action, tool_name, + model_used, policy_decision, approval_state, + outcome, details, timestamp + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NOW()) + """, + event.tenant_id, + event.agent_name, + event.action, + event.tool_name, + event.model_used, + json.dumps( + event.policy_decision.model_dump() if event.policy_decision else None + ), + event.approval_state, + event.outcome, + json.dumps(event.details) if event.details else None, + ) + + log.info( + "Audit event logged: %s/%s -> %s", + event.agent_name, + event.action, + event.outcome, + ) + return True + except Exception as e: + log.warning( + "Audit log failed for %s/%s: %s", + event.agent_name, + event.action, + e, + ) + return False + finally: + if own_conn and conn is not None: + await conn.close() \ No newline at end of file diff --git a/apps/ai/src/control_plane/policy.py b/apps/ai/src/control_plane/policy.py new file mode 100644 index 0000000..125d7ce --- /dev/null +++ b/apps/ai/src/control_plane/policy.py @@ -0,0 +1,99 @@ +"""Policy Engine — deterministic policy checks for agent actions. + +Evaluates data classification, allowed models, external-facing status, +tool permissions, and health state. No LLM calls. +""" +from __future__ import annotations + +import logging + +from src.schemas.control_plane import AgentRegistration, PolicyDecision + +log = logging.getLogger(__name__) + +# Restricted data classifications that block all agent actions +_RESTRICTED_CLASSIFICATIONS = {"restricted", "pii", "confidential"} + + +class PolicyEngine: + """Deterministic policy evaluation for agent actions.""" + + def evaluate( + self, + agent: AgentRegistration, + requested_tool: str | None = None, + ) -> PolicyDecision: + """Evaluate whether an agent action is permitted. + + Args: + agent: The registered agent attempting the action. + requested_tool: The specific tool being requested, if any. + + Returns: + PolicyDecision with approval state and constraints. + """ + approved_tools: list[str] = [] + requires_human_approval = False + blocked_reason: str | None = None + + # 1. Health check — unhealthy agents are blocked + if agent.health_status == "unhealthy": + blocked_reason = "agent_health_unhealthy" + log.warning( + "Policy block: agent %s is unhealthy", + agent.agent_name, + ) + return PolicyDecision( + data_classification=agent.data_classification, + allowed_model_classes=[], + requires_human_approval=True, + blocked_reason=blocked_reason, + approved_tools=[], + ) + + # 2. Data classification check — restricted data blocks + if agent.data_classification.lower() in _RESTRICTED_CLASSIFICATIONS: + blocked_reason = "data_classification_restricted" + log.warning( + "Policy block: agent %s has restricted classification: %s", + agent.agent_name, + agent.data_classification, + ) + return PolicyDecision( + data_classification=agent.data_classification, + allowed_model_classes=[], + requires_human_approval=True, + blocked_reason=blocked_reason, + approved_tools=[], + ) + + # 3. External-facing check — forces human approval + if agent.external_facing: + requires_human_approval = True + + # 4. Degraded health — forces review but not full block + if agent.health_status == "degraded": + requires_human_approval = True + blocked_reason = "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}" + 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, + approved_tools=approved_tools, + ) \ No newline at end of file diff --git a/apps/ai/src/control_plane/registry.py b/apps/ai/src/control_plane/registry.py new file mode 100644 index 0000000..958cc9f --- /dev/null +++ b/apps/ai/src/control_plane/registry.py @@ -0,0 +1,75 @@ +"""Agent Registry — typed registration and lookup. + +All agents must register through this control plane before they +can emit alerts, propose tools, or write MissionState fields. +""" +from __future__ import annotations + +import logging +import threading +from typing import Literal + +from src.schemas.control_plane import AgentRegistration + +log = logging.getLogger(__name__) + + +class ControlPlaneRegistry: + """Thread-safe registry for agent registrations.""" + + def __init__(self) -> None: + self._agents: dict[str, AgentRegistration] = {} + self._lock = threading.Lock() + + def register(self, registration: AgentRegistration) -> None: + """Register an agent. Replaces existing registration silently.""" + with self._lock: + self._agents[registration.agent_name] = registration + log.info( + "Agent registered: %s (domain=%s, tier=%s, external=%s)", + registration.agent_name, + registration.domain, + registration.escalation_tier, + registration.external_facing, + ) + + def get(self, agent_name: str) -> AgentRegistration | None: + """Look up an agent by name.""" + with self._lock: + return self._agents.get(agent_name) + + def is_registered(self, agent_name: str) -> bool: + """Check if an agent is registered.""" + with self._lock: + return agent_name in self._agents + + def is_action_allowed(self, agent_name: str, tool_name: str) -> bool: + """Check if the agent is allowed to execute a specific tool.""" + with self._lock: + reg = self._agents.get(agent_name) + if reg is None: + return False + return tool_name in reg.allowed_tools + + def list_agents(self) -> list[AgentRegistration]: + """Return all registered agents.""" + with self._lock: + return list(self._agents.values()) + + def set_health( + self, agent_name: str, status: Literal["healthy", "degraded", "unhealthy"] + ) -> None: + """Update an agent's health status.""" + with self._lock: + reg = self._agents.get(agent_name) + if reg is None: + log.warning("Cannot set health for unknown agent: %s", agent_name) + return + self._agents[agent_name] = reg.model_copy( + update={"health_status": status} + ) + log.info( + "Agent health updated: %s -> %s", + agent_name, + status, + ) \ No newline at end of file diff --git a/apps/ai/src/hitl/manager.py b/apps/ai/src/hitl/manager.py index dd90266..e7f6de6 100644 --- a/apps/ai/src/hitl/manager.py +++ b/apps/ai/src/hitl/manager.py @@ -1,18 +1,20 @@ """HITL Manager — 3-tier human-in-the-loop routing. -Tier 1 — AUTO: severity=info, confidence>0.85, seen before -Tier 2 — SLACK REVIEW: severity=warning, confidence 0.60-0.85, or new pattern -Tier 3 — HUMAN OVERRIDE: severity=critical, confidence<0.60, or investor update +Tier 1 — AUTO: severity=info, confidence>0.85 +Tier 2 — REVIEW: severity=warning, confidence<0.85, or new pattern +Tier 3 — APPROVE: severity=critical, confidence<0.60, or investor update + +Adds resolve_suggested_tools() to surface tools that match the routing tier. """ from __future__ import annotations -# Guardrail state mapping (NEW — additive) +from src.agents.tools import get_tools_for_patterns + GUARDRAIL_STATE_MAP = { "auto": "informational", "review": "advisory", "approve": "reviewable", } -# "blocked" is set explicitly by guardrail engine, not derived from HITL class HITLManager: @@ -75,3 +77,17 @@ def route_extended( is_new_pattern=is_new_pattern, is_investor_update=is_investor_update, ) + + def resolve_suggested_tools( + self, + triggered_patterns: list[str], + tier: str | None = None, + ) -> list[dict[str, str]]: + """Find registered tools matching triggered patterns and HITL tier.""" + candidates = get_tools_for_patterns(triggered_patterns) + if tier is not None: + candidates = [t for t in candidates if t.hitl_tier == tier] + return [ + {"name": t.name, "description": t.description, "hitl_tier": t.hitl_tier} + for t in candidates + ] diff --git a/apps/ai/src/integrations/slack.py b/apps/ai/src/integrations/slack.py index 1b86b01..19a06fc 100644 --- a/apps/ai/src/integrations/slack.py +++ b/apps/ai/src/integrations/slack.py @@ -423,10 +423,7 @@ async def handle_slack_message( return result -# ── Demo delivery (Mockoon + capture sidecar) ──────────────────── - -SLACK_DELIVERY_URL = os.environ.get("SLACK_DELIVERY_URL", "http://localhost:3001/slack/deliver") -_CAPTURE_URL = "http://localhost:3002" +# ── Guardian alert delivery ────────────────────────────────────── async def deliver_guardian_alert( @@ -435,23 +432,30 @@ async def deliver_guardian_alert( pattern_name: str, severity: str, ) -> Dict[str, Any]: - """Deliver guardian alert to Mockoon (Slack mock) + capture sidecar.""" - payload: Dict[str, Any] = { - "tenant_id": tenant_id, - "pattern_name": pattern_name, - "severity": severity, - "text": message, - "channel": "C_NOVAPULSE_DEMO", - } - async with httpx.AsyncClient(timeout=10) as client: - # Primary delivery (Mockoon → behaves like Slack) - try: - await client.post(SLACK_DELIVERY_URL, json=payload) - except Exception as e: - logger.warning("Mockoon delivery failed: %s", e) - # Capture sidecar (Playwright reads from here) - try: - await client.post(_CAPTURE_URL, json=payload) - except Exception: - pass # capture is best-effort — never block delivery - return {"ok": True, "channel": "C_NOVAPULSE_DEMO"} + """Deliver guardian alert to Slack via the SDK WebClient. + + Uses the shared ``SlackClient`` from slack_client.py which sends + the alert through the Slack API (``chat.postMessage``) instead of + the old Mockoon webhook mock. + + This function preserves the same signature for backward compatibility. + See ``SlackClient.send_guardian_alert()`` for details. + + Args: + tenant_id: Tenant identifier. + message: The alert body text. + pattern_name: Name of the detected anomaly/pattern. + severity: ``"info"`` | ``"warning"`` | ``"critical"``. + + Returns: + Dict with ``ok``, ``channel``, and optional ``ts`` / ``error``. + """ + from src.integrations.slack_client import SlackClient + + client = SlackClient() + return await client.send_guardian_alert( + tenant_id=tenant_id, + message=message, + pattern_name=pattern_name, + severity=severity, + ) diff --git a/apps/ai/src/integrations/slack_buttons.py b/apps/ai/src/integrations/slack_buttons.py index e0a9074..5fd9329 100644 --- a/apps/ai/src/integrations/slack_buttons.py +++ b/apps/ai/src/integrations/slack_buttons.py @@ -79,25 +79,52 @@ def _handle_log_decision(alert_id: str) -> ButtonResult: def _send_feedback_signal(alert_id: str, response_type: str, score: float) -> None: - """Send feedback signal to Reflector.""" + """Send feedback signal to Reflector. + + Completes the ACE (Alert → Collect → Evaluate) loop: + 1. Score the button response via Reflector → updates Trust Battery. + 2. Write strategy confidence delta to Graphiti. + 3. Write full playbook entry via Curator.update() with verification. + """ # Skip in test environments to avoid blocking on async operations import sys if "pytest" in sys.modules or hasattr(sys, "_pytestfixturefunction"): return + # Extract context from alert_id (format: "{tenant}-{domain}-{id}" or "{prefix}-{id}") + tenant_id = alert_id.split("-")[0] if "-" in alert_id else "default" + domain = alert_id.split("-")[1] if "-" in alert_id else "general" + + # Step 1: Score → Trust Battery update try: from src.agents.cofounder.reflector import score_from_button score_from_button(alert_id, response_type, score) except (ImportError, Exception): pass + # Step 2: Strategy confidence write → Graphiti try: from src.agents.cofounder.curator import update_strategy_confidence update_strategy_confidence( - tenant_id=alert_id.split("-")[0] if "-" in alert_id else "default", - domain=alert_id.split("-")[1] if "-" in alert_id else "general", + tenant_id=tenant_id, + domain=domain, feedback_type=response_type, score=score, ) + except (ImportError, Exception): + pass + + # Step 3: Full playbook write via Curator → completes ACE loop + # Uses curator.update() which writes a playbook entry + runs + # verification assertions (cohesion, drift, feedback trend). + try: + from src.agents.cofounder.curator import Curator + curator = Curator(tenant_id=tenant_id) + curator.update( + domain=domain, + strategy=f"hitl_{response_type}", + score_delta=score, + evidence_count=1, + ) except (ImportError, Exception): pass \ No newline at end of file diff --git a/apps/ai/src/integrations/slack_client.py b/apps/ai/src/integrations/slack_client.py index d98eb97..e461bef 100644 --- a/apps/ai/src/integrations/slack_client.py +++ b/apps/ai/src/integrations/slack_client.py @@ -1,16 +1,72 @@ +import json +import logging import os from slack_sdk import WebClient from slack_sdk.socket_mode import SocketModeClient from slack_sdk.socket_mode.request import SocketModeRequest from slack_sdk.socket_mode.response import SocketModeResponse +logger = logging.getLogger(__name__) + + class SlackClient: def __init__(self): self.client = WebClient(token=os.getenv("SLACK_BOT_TOKEN")) self.socket_client = SocketModeClient( app_token=os.getenv("SLACK_APP_TOKEN"), - web_client=self.client + web_client=self.client, ) + self._socket_listener_running = False + + async def start_socket_mode_listener(self) -> SocketModeClient: + """Start listening for Slack interactive events via Socket Mode. + + Registers a listener on the existing SocketModeClient that captures + interactive button payloads and routes them through the shared + ``route_slack_button()`` handler from slack_buttons.py. + + Returns: + The connected SocketModeClient instance, or the existing one + if already running (idempotent). + + Environment: + SLACK_APP_TOKEN — Socket Mode app-level token (starts with ``xapp-``) + SLACK_BOT_TOKEN — Bot token used by the WebClient (already set in __init__) + """ + if self._socket_listener_running: + logger.info("Socket Mode listener already running — skipping") + return self.socket_client + + from src.integrations.slack_buttons import route_slack_button + + def _process(client: SocketModeClient, req: SocketModeRequest) -> None: + """Synchronous callback for incoming Socket Mode requests. + + ``route_slack_button()`` is synchronous (no IO), and the non-aiohttp + ``SocketModeClient`` uses a sync WebSocket client, so this callback + is intentionally sync to avoid coroutine mismatches with ``ack()``. + """ + try: + if req.type == "interactive": + payload = json.loads(req.payload) + result = route_slack_button(payload) + if result.error: + logger.warning("Button route returned error", extra={"error": result.error}) + # Always ack so Slack doesn't retry + req.ack() + except Exception: + logger.exception("Unhandled error in Socket Mode listener") + # Still ack even on failure to prevent Slack retries + try: + req.ack() + except Exception: + pass + + self.socket_client.socket_mode_request_listeners.append(_process) + await self.socket_client.connect() + self._socket_listener_running = True + logger.info("Socket Mode listener connected and running") + return self.socket_client async def open_decision_modal(self, trigger_id: str): """Open the decision logging modal""" @@ -140,4 +196,46 @@ def get_channel_id_by_name(self, channel_name: str) -> str | None: return channel.get("id") return None except Exception: - return None \ No newline at end of file + return None + + async def send_guardian_alert( + self, + tenant_id: str, + message: str, + pattern_name: str, + severity: str, + ) -> dict: + """Send a guardian alert to the guardian-alerts Slack channel via WebClient. + + Replaces the old Mockoon-based ``deliver_guardian_alert()`` with a real + Slack API call. The alert is sent as a formatted message with severity + prefix and pattern context. + + Args: + tenant_id: Tenant identifier (included in message context). + message: The alert body text. + pattern_name: Name of the detected anomaly/pattern. + severity: One of ``"info"``, ``"warning"``, ``"critical"``. + + Returns: + Dict with ``ok``, ``channel``, and optionally ``ts`` (message timestamp) + or ``error``. + + Environment: + SLACK_GUARDIAN_CHANNEL — Channel to post alerts to (default: ``#guardian-alerts``). + """ + channel = os.getenv("SLACK_GUARDIAN_CHANNEL", "#guardian-alerts") + try: + response = self.client.chat_postMessage( + channel=channel, + text=f"[{severity.upper()}] *{pattern_name}* (tenant: {tenant_id}): {message}", + ) + ts = response.get("ts", "") + logger.info( + "Guardian alert sent", + extra={"channel": channel, "pattern": pattern_name, "severity": severity, "ts": ts}, + ) + return {"ok": True, "channel": channel, "ts": ts} + except Exception as e: + logger.error("Failed to send guardian alert", extra={"error": str(e)}) + return {"ok": False, "channel": channel, "error": str(e)} \ No newline at end of file diff --git a/apps/ai/src/risk/__init__.py b/apps/ai/src/risk/__init__.py new file mode 100644 index 0000000..e8fbbcd --- /dev/null +++ b/apps/ai/src/risk/__init__.py @@ -0,0 +1,8 @@ +"""Risk scanning — prompt and output risk guards for sensitive workflows.""" +from src.risk.prompt_risk import scan_prompt +from src.risk.output_risk import scan_output + +__all__ = [ + "scan_prompt", + "scan_output", +] diff --git a/apps/ai/src/risk/output_risk.py b/apps/ai/src/risk/output_risk.py new file mode 100644 index 0000000..21b469c --- /dev/null +++ b/apps/ai/src/risk/output_risk.py @@ -0,0 +1,126 @@ +"""Output Risk Scanner — deterministic post-generation scan. + +Scans generated drafts for unsupported claims, promises, +pricing commitments, investor misstatements, or missing +approval state before any send action is permitted. +""" +from __future__ import annotations + +import logging +import re + +from src.schemas.control_plane import RiskFlag, RiskScanResult, RiskSeverity + +log = logging.getLogger(__name__) + +# Unsupported claim patterns — specific numbers presented as projections +_UNSUPPORTED_CLAIMS: list[tuple[str, str, RiskSeverity]] = [ + ("OC001", r"(?i)(growing\s+\d+%\s+(MoM|YoY|month|year)|(\d+[-×x])\d+%\s+growth)", RiskSeverity.HIGH), + ("OC002", r"(?i)(on\s+track\s+to\s+(hit|reach|achieve)\s+\$?[\d,.]+[kKmMbB]?)", RiskSeverity.HIGH), +] + +# Promise/commitment patterns +_PROMISE_PATTERNS: list[tuple[str, str, RiskSeverity]] = [ + ("OC003", r"(?i)(guarantee|guaranteed|promise|promised|ensure|assure|will\s+definitely|certainly)", RiskSeverity.HIGH), + ("OC004", r"(?i)(we\s+(will|shall)\s+(always|never|absolutely|unconditionally))", RiskSeverity.MEDIUM), +] + +# Pricing commitment patterns +_PRICING_PATTERNS: list[tuple[str, str, RiskSeverity]] = [ + ("OC005", r"(?i)(price\s+(freeze|lock|guarantee)|no\s+price\s+increase|flat\s+pricing)", RiskSeverity.HIGH), + ("OC006", r"(?i)(discount\s+(of\s+)?\d+%|reduced\s+(to|by)\s+\$?[\d,.]+)", RiskSeverity.MEDIUM), +] + +# Investor misstatement patterns +_INVESTOR_MISSTATEMENTS: list[tuple[str, str, RiskSeverity]] = [ + ("OC007", r"(?i)(outpacing\s+(competitors|market|industry)|disrupting\s+the\s+\w+\s+industry)", RiskSeverity.MEDIUM), + ("OC008", r"(?i)(comparable\s+to\s+[A-Z][a-z]+(\s+[A-Z][a-z]+)?\s+(at\s+)?(our|similar)\s+stage)", RiskSeverity.MEDIUM), +] + +# Missing approval state — output should indicate it's a draft +_MISSING_APPROVAL = [ + ("OC009", r"(?i)(draft|for\s+review|pending\s+approval|needs\s+review|not\s+final)", RiskSeverity.LOW), +] + +_ALL_OUTPUT_PATTERNS = _UNSUPPORTED_CLAIMS + _PROMISE_PATTERNS + _PRICING_PATTERNS + _INVESTOR_MISSTATEMENTS + _MISSING_APPROVAL + + +def scan_output(text: str, context: str | None = None) -> RiskScanResult: + """Scan generated output text for risk flags before release. + + Args: + text: The generated draft or output text to scan. + context: Optional context hint (e.g. "investor_update", "customer_email"). + + Returns: + RiskScanResult with any flags detected. + """ + flags: list[RiskFlag] = [] + highest_severity: RiskSeverity = RiskSeverity.LOW + has_draft_disclaimer = False + + for rule_id, pattern, severity in _MISSING_APPROVAL: + if re.search(pattern, text): + has_draft_disclaimer = True + break + + 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, + )) + + # If context is investor-facing and no draft disclaimer found, add a warning + if context in ("investor_update", "customer_email") and not has_draft_disclaimer: + flags.append(RiskFlag( + rule_id="OC009", + description="Output is missing draft/pending-review disclaimer", + severity=RiskSeverity.LOW, + matched_text="(no draft disclaimer found)", + )) + + if flags: + status = "flag" + highest_severity = max((f.severity for f in flags), key=_severity_rank) + if highest_severity == RiskSeverity.HIGH: + recommended_action = "block" + elif highest_severity == RiskSeverity.MEDIUM: + recommended_action = "review" + else: + recommended_action = "proceed" + else: + status = "pass" + recommended_action = "proceed" + + return RiskScanResult( + status=status, + flags=flags, + severity=highest_severity, + recommended_action=recommended_action, + ) + + +def _severity_rank(s: RiskSeverity) -> int: + return {"low": 0, "medium": 1, "high": 2}.get(s.value, 0) + + +def _get_output_rule_description(rule_id: str) -> str: + descriptions = { + "OC001": "Output contains unsupported growth claim", + "OC002": "Output contains unsupported revenue projection", + "OC003": "Output contains guarantee or promise", + "OC004": "Output contains absolute commitment language", + "OC005": "Output contains pricing commitment", + "OC006": "Output contains discount promise", + "OC007": "Output contains misleading competitive claim", + "OC008": "Output contains false comparable claim", + "OC009": "Output is missing draft/pending-review disclaimer", + } + return descriptions.get(rule_id, f"Unknown rule: {rule_id}") diff --git a/apps/ai/src/risk/prompt_risk.py b/apps/ai/src/risk/prompt_risk.py new file mode 100644 index 0000000..e8e6943 --- /dev/null +++ b/apps/ai/src/risk/prompt_risk.py @@ -0,0 +1,108 @@ +"""Prompt Risk Scanner — deterministic pre-generation scan. + +Scans assembled prompts for restricted content, customer PII, +investor-sensitive phrasing, and disallowed external-send actions +before any LLM invocation. +""" +from __future__ import annotations + +import logging +import re + +from src.schemas.control_plane import RiskFlag, RiskScanResult, RiskSeverity + +log = logging.getLogger(__name__) + +# Restricted tokens that should never appear in prompts sent to LLMs +_RESTRICTED_PATTERNS: list[tuple[str, str, RiskSeverity]] = [ + ("R001", r"(?i)(api[_-]?key|secret[_-]?key|password|auth[_-]?token)\s*[:=]\s*['\"]?\w{16,}", RiskSeverity.HIGH), + ("R002", r"(?i)(sk-[a-zA-Z0-9]{20,}|pk-[a-zA-Z0-9]{20,})", RiskSeverity.HIGH), +] + +# Customer-identifying patterns +_CUSTOMER_PII_PATTERNS: list[tuple[str, str, RiskSeverity]] = [ + ("P001", r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", RiskSeverity.MEDIUM), + ("P002", r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", RiskSeverity.HIGH), +] + +# Investor-sensitive phrasing — claims that need disclaimers +_INVESTOR_SENSITIVE_KEYWORDS: list[tuple[str, str, RiskSeverity]] = [ + ("I001", r"(?i)(projected|forecast|guidance|will\s+(grow|hit|reach|achieve|deliver))", RiskSeverity.MEDIUM), + ("I002", r"(?i)(revenue\s+(run[-\s]?rate|projection|target)|mrr\s+projection)", RiskSeverity.MEDIUM), + ("I003", r"(?i)(ipo|acquisition\s+offer|exit\s+strategy|series\s+[abcde])", RiskSeverity.MEDIUM), +] + +# Disallowed external-send action patterns +_EXTERNAL_SEND_PATTERNS: list[tuple[str, str, RiskSeverity]] = [ + ("E001", r"(?i)(send\s+(this|the|an?\s+(email|update|draft))|email\s+(to|the|this))", RiskSeverity.HIGH), + ("E002", r"(?i)(post\s+(to|on|this)|publish\s+(this|the)|share\s+(with|externally))", RiskSeverity.HIGH), +] + +_ALL_PATTERNS = _RESTRICTED_PATTERNS + _CUSTOMER_PII_PATTERNS + _INVESTOR_SENSITIVE_KEYWORDS + _EXTERNAL_SEND_PATTERNS + + +def scan_prompt(text: str, context: str | None = None) -> RiskScanResult: + """Scan prompt text for risk flags before LLM invocation. + + Args: + text: The prompt or assembled input text to scan. + context: Optional context hint (e.g. "investor_update", "customer_email"). + + Returns: + RiskScanResult with any flags detected. + """ + flags: list[RiskFlag] = [] + highest_severity: RiskSeverity = RiskSeverity.LOW + + 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, + )) + + if flags: + status = "flag" + highest_severity = max((f.severity for f in flags), key=_severity_rank) + if highest_severity == RiskSeverity.HIGH or context in ("investor_update", "customer_email"): + recommended_action = "block" + elif highest_severity == RiskSeverity.MEDIUM: + recommended_action = "review" + else: + recommended_action = "proceed" + else: + status = "pass" + recommended_action = "proceed" + + return RiskScanResult( + status=status, + flags=flags, + severity=highest_severity, + recommended_action=recommended_action, + ) + + +def _severity_rank(s: RiskSeverity) -> int: + return {"low": 0, "medium": 1, "high": 2}.get(s.value, 0) + + +def _get_rule_description(rule_id: str) -> str: + descriptions = { + "R001": "Prompt contains credentials or secrets", + "R002": "Prompt contains API key pattern", + "P001": "Prompt contains email address", + "P002": "Prompt contains phone number pattern", + "I001": "Prompt contains forward-looking statement", + "I002": "Prompt contains revenue projection", + "I003": "Prompt contains exit/valuation language", + "E001": "Prompt contains send action instruction", + "E002": "Prompt contains publish/share action instruction", + } + return descriptions.get(rule_id, f"Unknown rule: {rule_id}") diff --git a/apps/ai/src/schemas/__init__.py b/apps/ai/src/schemas/__init__.py index 52ceeb0..aee9fc6 100644 --- a/apps/ai/src/schemas/__init__.py +++ b/apps/ai/src/schemas/__init__.py @@ -1,7 +1,7 @@ """ Pydantic schemas for TrackGuard v4.2 Phase 3 desk agents. -Export all desk result types for easy importing. +Export all desk result types and control plane schemas for easy importing. """ from src.schemas.desk_results import ( HitlRisk, @@ -14,6 +14,15 @@ DeskResult, ) +from src.schemas.control_plane import ( + PolicyDecision, + RiskScanResult, + RiskFlag, + RiskSeverity, + AgentRegistration, + AuditEvent, +) + __all__ = [ "HitlRisk", "FinanceTaskResult", @@ -23,4 +32,10 @@ "ITRiskAlert", "KnowledgeManagerResult", "DeskResult", + "PolicyDecision", + "RiskScanResult", + "RiskFlag", + "RiskSeverity", + "AgentRegistration", + "AuditEvent", ] diff --git a/apps/ai/src/schemas/control_plane.py b/apps/ai/src/schemas/control_plane.py new file mode 100644 index 0000000..8513331 --- /dev/null +++ b/apps/ai/src/schemas/control_plane.py @@ -0,0 +1,105 @@ +"""Control Plane schemas for agent registry, policy, risk, and audit.""" +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from pydantic import BaseModel, Field +from typing import Literal + + +class RiskSeverity(str, Enum): + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + + +class RiskFlag(BaseModel): + """A single risk flag detected during prompt or output scanning.""" + rule_id: str + description: str + severity: RiskSeverity + matched_text: str + + +class RiskScanResult(BaseModel): + """Result of a prompt or output risk scan. + + Attributes: + status: "pass" (no flags), "flag" (issues found), "error" + flags: List of RiskFlag objects + severity: Overall severity ("low", "medium", "high") + recommended_action: "proceed", "review", "block" + """ + status: Literal["pass", "flag", "error"] + flags: list[RiskFlag] = Field(default_factory=list) + severity: RiskSeverity = RiskSeverity.LOW + recommended_action: Literal["proceed", "review", "block"] = "proceed" + + +class PolicyDecision(BaseModel): + """Outcome of a policy evaluation for an agent action. + + Attributes: + data_classification: Classification of data involved ("internal", "external_investor", "external_customer", "restricted") + allowed_model_classes: Which model classes are permitted (e.g. ["gpt-4o"], ["gpt-4o-mini"]) + requires_human_approval: Whether HITL approval is required + blocked_reason: If blocked, the reason string; None if allowed + approved_tools: List of tool names approved for this action + """ + data_classification: str + allowed_model_classes: list[str] + requires_human_approval: bool + blocked_reason: str | None = None + approved_tools: list[str] = Field(default_factory=list) + + +class AuditEvent(BaseModel): + """Audit log entry for a control-plane-gated agent action. + + Attributes: + agent_name: Name of the agent that performed the action + action: Type of action ("tool_execution", "mission_state_write", "alert_emit", "llm_invoke") + tool_name: Name of the tool if action is tool_execution + model_used: Model class used for LLM call + policy_decision: The PolicyDecision that governed this action + approval_state: "auto", "review", "approve", "blocked" + outcome: "completed", "blocked", "failed" + tenant_id: Tenant identifier + timestamp: When the event occurred + details: Optional additional context + """ + agent_name: str + action: Literal["tool_execution", "mission_state_write", "alert_emit", "llm_invoke"] + tool_name: str | None = None + model_used: str | None = None + policy_decision: PolicyDecision | None = None + approval_state: str | None = None + outcome: Literal["completed", "blocked", "failed"] + tenant_id: str + timestamp: datetime | None = None + details: dict | None = None + + +class AgentRegistration(BaseModel): + """Registration record for an agent in the control plane. + + Attributes: + agent_name: Unique agent name + role: Human-readable role description + domain: Domain literal ("finance", "bi", "ops", "cofounder", "correlation") + allowed_tools: List of tool names this agent may execute + allowed_models: List of model classes this agent may use + escalation_tier: Default HITL tier ("auto", "review", "approve", "blocked") + external_facing: Whether this agent produces external-facing outputs + data_classification: Default data classification + health_status: "healthy", "degraded", "unhealthy" + """ + agent_name: str + role: str + domain: Literal["finance", "bi", "ops", "cofounder", "correlation"] + allowed_tools: list[str] + allowed_models: list[str] + escalation_tier: Literal["auto", "review", "approve", "blocked"] + external_facing: bool = False + data_classification: str = "internal" + health_status: Literal["healthy", "degraded", "unhealthy"] = "healthy" diff --git a/apps/ai/src/schemas/guardian.py b/apps/ai/src/schemas/guardian.py index 99e658b..9b34d86 100644 --- a/apps/ai/src/schemas/guardian.py +++ b/apps/ai/src/schemas/guardian.py @@ -29,6 +29,22 @@ from typing import Literal +class AlertLineage(BaseModel): + """Provenance chain for a Guardian alert — links pattern → data → mission. + + All fields are code-generated, never LLM output. pattern_id and owner_agent + come from the watchlist/blindspot definition. source_metrics come from the + pattern definition (blindspot.id). mission_context comes from MissionState + fields. raise_timeline_risk is a human-readable escalation urgency string. + """ + pattern_id: str + source_metrics: list[str] + mission_context: list[str] + raise_timeline_risk: str + suggested_tool_ids: list[str] + owner_agent: str + + class AlertDecision(BaseModel): """ Output contract for Guardian's cognitive decision. @@ -89,6 +105,7 @@ class GuardianMessage(BaseModel): urgency_horizon: Literal["today", "this_week", "this_month", "this_quarter"] one_action: str injected_numbers: list[str] = Field(default_factory=list) + lineage: AlertLineage @field_validator("insight") @classmethod diff --git a/apps/ai/src/session/brief_generator.py b/apps/ai/src/session/brief_generator.py new file mode 100644 index 0000000..ff83083 --- /dev/null +++ b/apps/ai/src/session/brief_generator.py @@ -0,0 +1,49 @@ +"""Brief generator — 2-sentence plain-English business summary after MissionState updates.""" +from __future__ import annotations + +import logging +from src.config.llm import chat_completion +from src.session.mission_state import get_mission_state, update_mission_state, MissionState + +log = logging.getLogger(__name__) + +_BRIEF_TEMPLATE = ( + "Write 2 plain-English sentences summarising this business state. " + "Runway: {runway_days}d | Burn alert: {burn_alert} | " + "Churn rate: {churn_rate} | Active alerts: {active_alerts} | " + "MRR trend: {mrr_trend} | Trust score: {trust_score}. " + "No jargon. Founder reads this first thing. Be direct." +) + +async def generate_prepared_brief(tenant_id: str) -> str | None: + """Load MissionState, generate 2-sentence brief, persist it, return it.""" + state = await get_mission_state(tenant_id) + prompt = _BRIEF_TEMPLATE.format( + runway_days=state.runway_days or "?", + burn_alert=state.burn_alert, + churn_rate=state.churn_rate or "?", + active_alerts=state.active_alerts or "none", + mrr_trend=state.mrr_trend or "?", + trust_score=state.trust_score or "?", + ) + try: + # chat_completion is synchronous + brief = chat_completion( + messages=[ + {"role": "system", "content": "You are a concise business briefing assistant. Output exactly 2 sentences. No preamble."}, + {"role": "user", "content": prompt}, + ], + max_tokens=80, + temperature=0.3, + ) + if not brief: + log.warning("Empty brief returned for tenant %s", tenant_id) + return None + state.prepared_brief = brief.strip() + state.last_updated_by = "brief_generator" + await update_mission_state(state, generate_brief=False) + log.info("Prepared brief generated for tenant %s: %s", tenant_id, brief[:80]) + return brief.strip() + except Exception: + log.exception("Failed to generate prepared brief for tenant %s", tenant_id) + return None diff --git a/apps/ai/src/session/mission_state.py b/apps/ai/src/session/mission_state.py index 99148d3..a6f9521 100644 --- a/apps/ai/src/session/mission_state.py +++ b/apps/ai/src/session/mission_state.py @@ -10,6 +10,7 @@ """ from __future__ import annotations +import json import os import logging from dataclasses import dataclass @@ -88,6 +89,19 @@ class MissionState: guardrail_blocking: bool = False investor_facing_alert: bool = False + # ── Cognitive offloading fields ────────────────────────────────── + prepared_brief: str | None = None # LLM-generated brief for founder context + pending_decisions: list[dict] | None = None # JSONB array of pending founder decisions + last_updated_by: str | None = None # which agent/specialist last wrote to MissionState + + # ── Explainability fields (Paperclip-inspired) ────────────────── + last_update_reason: str | None = None # why this write happened + last_changed_fields: list[str] | None = None # which fields were modified + active_agent_roles: list[str] | None = None # derived from authority manifest + + # ── Policy state ───────────────────────────────────────────────── + policy_state: dict | None = None # last PolicyDecision dict for dashboard display + async def get_mission_state(tenant_id: str) -> MissionState: """Get MissionState from database. @@ -110,7 +124,10 @@ async def get_mission_state(tenant_id: str) -> MissionState: burn_multiple, effective_runway_days, working_capital_ratio, npv_last_decision, wacc_estimate, last_approval_tier, last_reversible, active_authority_limit, guardrail_override_reason, guardrail_risk_type, - guardrail_blocking, investor_facing_alert + guardrail_blocking, investor_facing_alert, + prepared_brief, pending_decisions, last_updated_by, + last_update_reason, last_changed_fields, active_agent_roles, + policy_state FROM mission_states WHERE tenant_id = $1 ORDER BY timestamp DESC @@ -148,6 +165,13 @@ async def get_mission_state(tenant_id: str) -> MissionState: guardrail_risk_type=row["guardrail_risk_type"], guardrail_blocking=row["guardrail_blocking"], investor_facing_alert=row["investor_facing_alert"], + prepared_brief=row["prepared_brief"], + pending_decisions=row["pending_decisions"], + last_updated_by=row["last_updated_by"], + last_update_reason=row["last_update_reason"], + last_changed_fields=row["last_changed_fields"], + active_agent_roles=row["active_agent_roles"], + policy_state=row["policy_state"], ) except Exception as e: log.warning(f"MissionState lookup failed for {tenant_id}: {e}") @@ -155,17 +179,23 @@ async def get_mission_state(tenant_id: str) -> MissionState: return MissionState(tenant_id=tenant_id) -async def update_mission_state(state: MissionState) -> bool: +async def update_mission_state(state: MissionState, generate_brief: bool = True, update_reason: str | None = None, changed_fields: list[str] | None = None) -> bool: """Update MissionState in database atomically. Per PRD Section 11: Updated atomically. Args: state: MissionState to persist + generate_brief: Whether to auto-generate prepared_brief if missing + update_reason: Why this write happened + changed_fields: Which fields were modified in this update Returns: True if successful, False otherwise """ + if generate_brief: + state.last_update_reason = update_reason + state.last_changed_fields = changed_fields try: conn = await asyncpg.connect(DATABASE_URL) await conn.execute( @@ -177,10 +207,14 @@ async def update_mission_state(state: MissionState) -> bool: burn_multiple, effective_runway_days, working_capital_ratio, npv_last_decision, wacc_estimate, last_approval_tier, last_reversible, active_authority_limit, guardrail_override_reason, guardrail_risk_type, - guardrail_blocking, investor_facing_alert, created_at + guardrail_blocking, investor_facing_alert, created_at, + prepared_brief, pending_decisions, last_updated_by, + last_update_reason, last_changed_fields, active_agent_roles, + policy_state ) VALUES ($1, NOW(), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, - $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, NOW()) + $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, NOW(), + $26, $27, $28, $29, $30, $31, $32) ON CONFLICT (tenant_id) DO UPDATE SET timestamp = NOW(), runway_days = EXCLUDED.runway_days, @@ -206,7 +240,14 @@ async def update_mission_state(state: MissionState) -> bool: guardrail_override_reason = EXCLUDED.guardrail_override_reason, guardrail_risk_type = EXCLUDED.guardrail_risk_type, guardrail_blocking = EXCLUDED.guardrail_blocking, - investor_facing_alert = EXCLUDED.investor_facing_alert + investor_facing_alert = EXCLUDED.investor_facing_alert, + prepared_brief = EXCLUDED.prepared_brief, + pending_decisions = EXCLUDED.pending_decisions, + last_updated_by = EXCLUDED.last_updated_by, + last_update_reason = EXCLUDED.last_update_reason, + last_changed_fields = EXCLUDED.last_changed_fields, + active_agent_roles = EXCLUDED.active_agent_roles, + policy_state = EXCLUDED.policy_state """, state.tenant_id, state.runway_days, @@ -233,8 +274,21 @@ async def update_mission_state(state: MissionState) -> bool: state.guardrail_risk_type, state.guardrail_blocking, state.investor_facing_alert, + 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") log.info(f"MissionState updated for tenant: {state.tenant_id}") return True except Exception as e: diff --git a/apps/ai/tests/unit/test_control_plane.py b/apps/ai/tests/unit/test_control_plane.py new file mode 100644 index 0000000..64d01d6 --- /dev/null +++ b/apps/ai/tests/unit/test_control_plane.py @@ -0,0 +1,389 @@ +"""Control Plane tests — pure Python, zero infra.""" +import pytest +from pydantic import ValidationError + +from src.schemas.control_plane import ( + PolicyDecision, + RiskScanResult, + AgentRegistration, + AuditEvent, + RiskFlag, + RiskSeverity, +) +from src.control_plane.registry import ControlPlaneRegistry +from src.control_plane.policy import PolicyEngine +from src.agents.authority_manifest import AgentAuthority + + +class TestPolicyDecisionSchema: + def test_valid_policy_decision(self): + d = PolicyDecision( + data_classification="internal", + allowed_model_classes=["gpt-4o"], + requires_human_approval=False, + blocked_reason=None, + approved_tools=["flag_churn_risk_customer"], + ) + assert d.data_classification == "internal" + assert d.allowed_model_classes == ["gpt-4o"] + assert d.requires_human_approval is False + assert d.approved_tools == ["flag_churn_risk_customer"] + + def test_external_facing_forces_approval(self): + d = PolicyDecision( + data_classification="external_investor", + allowed_model_classes=["gpt-4o"], + requires_human_approval=True, + blocked_reason="external_facing_output", + approved_tools=["draft_investor_update"], + ) + assert d.requires_human_approval is True + assert d.blocked_reason == "external_facing_output" + + def test_blocked_decision(self): + d = PolicyDecision( + data_classification="restricted", + allowed_model_classes=[], + requires_human_approval=True, + blocked_reason="data_classification_restricted", + approved_tools=[], + ) + assert d.blocked_reason == "data_classification_restricted" + assert d.approved_tools == [] + + +class TestRiskScanResultSchema: + def test_clean_scan(self): + r = RiskScanResult( + status="pass", + flags=[], + severity="low", + recommended_action="proceed", + ) + assert r.status == "pass" + assert r.severity == "low" + + def test_flagged_scan(self): + r = RiskScanResult( + status="flag", + flags=[ + RiskFlag( + rule_id="R001", + description="Contains unsupported growth claim", + severity="high", + matched_text="growing 300% MoM", + ) + ], + severity="high", + recommended_action="block", + ) + assert r.status == "flag" + assert len(r.flags) == 1 + assert r.flags[0].rule_id == "R001" + assert r.recommended_action == "block" + + +class TestAgentRegistrationSchema: + def test_valid_registration(self): + a = AgentRegistration( + agent_name="Sarthi · Finance", + role="Finance specialist", + domain="finance", + allowed_tools=["pause_failed_payment_retry"], + allowed_models=["gpt-4o-mini"], + escalation_tier="review", + external_facing=False, + data_classification="internal", + health_status="healthy", + ) + assert a.agent_name == "Sarthi · Finance" + assert a.health_status == "healthy" + + def test_external_facing_registration(self): + a = AgentRegistration( + agent_name="Sarthi", + role="manager/cofounder", + domain="cofounder", + allowed_tools=["draft_investor_update"], + allowed_models=["gpt-4o"], + escalation_tier="approve", + external_facing=True, + data_classification="external_investor", + health_status="healthy", + ) + assert a.external_facing is True + assert a.escalation_tier == "approve" + + def test_invalid_domain(self): + with pytest.raises(ValidationError): + AgentRegistration( + agent_name="BadAgent", + role="hacker", + domain="unknown", + allowed_tools=[], + allowed_models=[], + escalation_tier="auto", + external_facing=False, + data_classification="internal", + health_status="healthy", + ) + + +class TestAuditEventSchema: + def test_audit_event_creation(self): + e = AuditEvent( + agent_name="Sarthi · Finance", + action="tool_execution", + tool_name="pause_failed_payment_retry", + model_used="gpt-4o-mini", + policy_decision=PolicyDecision( + data_classification="internal", + allowed_model_classes=["gpt-4o-mini"], + requires_human_approval=False, + blocked_reason=None, + approved_tools=["pause_failed_payment_retry"], + ), + approval_state="auto", + outcome="completed", + tenant_id="test-tenant", + ) + assert e.agent_name == "Sarthi · Finance" + assert e.outcome == "completed" + + +class TestControlPlaneRegistry: + def test_register_and_get_agent(self): + registry = ControlPlaneRegistry() + reg = AgentRegistration( + agent_name="TestAgent", + role="tester", + domain="ops", + allowed_tools=[], + allowed_models=["gpt-4o-mini"], + escalation_tier="auto", + external_facing=False, + data_classification="internal", + health_status="healthy", + ) + registry.register(reg) + fetched = registry.get("TestAgent") + assert fetched is not None + assert fetched.agent_name == "TestAgent" + + def test_unregistered_agent_returns_none(self): + registry = ControlPlaneRegistry() + assert registry.get("GhostAgent") is None + + def test_agent_must_register_with_control_plane(self): + registry = ControlPlaneRegistry() + reg = AgentRegistration( + agent_name="Sarthi · Finance", + role="Finance specialist", + domain="finance", + allowed_tools=["pause_failed_payment_retry"], + allowed_models=["gpt-4o-mini"], + escalation_tier="review", + external_facing=False, + data_classification="internal", + health_status="healthy", + ) + registry.register(reg) + + assert registry.is_registered("Sarthi · Finance") is True + assert registry.is_registered("GhostAgent") is False + assert registry.is_action_allowed("Sarthi · Finance", "pause_failed_payment_retry") is True + assert registry.is_action_allowed("Sarthi · Finance", "draft_investor_update") is False + + def test_list_agents(self): + registry = ControlPlaneRegistry() + reg = AgentRegistration( + agent_name="ListTest", + role="tester", + domain="ops", + allowed_tools=[], + allowed_models=[], + escalation_tier="auto", + external_facing=False, + data_classification="internal", + health_status="healthy", + ) + registry.register(reg) + agents = registry.list_agents() + assert any(a.agent_name == "ListTest" for a in agents) + + def test_health_status_toggle(self): + registry = ControlPlaneRegistry() + reg = AgentRegistration( + agent_name="HealthTest", + role="tester", + domain="ops", + allowed_tools=[], + allowed_models=[], + escalation_tier="auto", + external_facing=False, + data_classification="internal", + health_status="healthy", + ) + registry.register(reg) + registry.set_health("HealthTest", "degraded") + assert registry.get("HealthTest").health_status == "degraded" + registry.set_health("HealthTest", "unhealthy") + assert registry.get("HealthTest").health_status == "unhealthy" + + +class TestPolicyEngine: + def test_internal_policy_auto(self): + engine = PolicyEngine() + reg = AgentRegistration( + agent_name="OPS Agent", + role="ops", + domain="ops", + allowed_tools=["flag_churn_risk_customer"], + allowed_models=["gpt-4o-mini"], + escalation_tier="auto", + external_facing=False, + data_classification="internal", + health_status="healthy", + ) + decision = engine.evaluate(reg, requested_tool="flag_churn_risk_customer") + assert decision.requires_human_approval is False + assert "gpt-4o-mini" in decision.allowed_model_classes + + def test_external_facing_outputs_force_hitl_approve(self): + engine = PolicyEngine() + reg = AgentRegistration( + agent_name="Sarthi", + role="cofounder", + domain="cofounder", + allowed_tools=["draft_investor_update"], + allowed_models=["gpt-4o"], + escalation_tier="approve", + external_facing=True, + data_classification="external_investor", + health_status="healthy", + ) + decision = engine.evaluate(reg, requested_tool="draft_investor_update") + assert decision.requires_human_approval is True + assert decision.data_classification == "external_investor" + + def test_disallowed_tool_blocked(self): + engine = PolicyEngine() + reg = AgentRegistration( + agent_name="OPS Agent", + role="ops", + domain="ops", + allowed_tools=["flag_churn_risk_customer"], + allowed_models=["gpt-4o-mini"], + escalation_tier="auto", + external_facing=False, + data_classification="internal", + health_status="healthy", + ) + decision = engine.evaluate(reg, requested_tool="draft_investor_update") + assert "draft_investor_update" not in decision.approved_tools + assert decision.blocked_reason is not None + + def test_unhealthy_agent_blocked(self): + engine = PolicyEngine() + reg = AgentRegistration( + agent_name="BrokenAgent", + role="ops", + domain="ops", + allowed_tools=["flag_churn_risk_customer"], + allowed_models=["gpt-4o-mini"], + escalation_tier="auto", + external_facing=False, + data_classification="internal", + health_status="unhealthy", + ) + decision = engine.evaluate(reg, requested_tool="flag_churn_risk_customer") + assert decision.requires_human_approval is True + assert "unhealthy" in (decision.blocked_reason or "") + + def test_restricted_data_classification_blocks(self): + engine = PolicyEngine() + reg = AgentRegistration( + agent_name="RestrictedAgent", + role="ops", + domain="ops", + allowed_tools=["flag_churn_risk_customer"], + allowed_models=["gpt-4o-mini"], + escalation_tier="auto", + external_facing=False, + data_classification="restricted", + health_status="healthy", + ) + decision = engine.evaluate(reg, requested_tool="flag_churn_risk_customer") + assert "restricted" in (decision.blocked_reason or "") + + +class TestExternalFacingHITLEnforcement: + def test_external_facing_workflow_always_approve(self): + from src.hitl.manager import HITLManager + m = HITLManager() + result = m.route( + severity="info", + confidence=0.99, + is_investor_update=True, + ) + assert result == "approve" + + def test_external_facing_extended_always_approve(self): + from src.hitl.manager import HITLManager + m = HITLManager() + result = m.route_extended( + severity="info", + confidence=0.99, + approval_required=True, + ) + assert result == "approve" + + def test_external_facing_overrides_risk_tolerance(self): + from src.hitl.manager import HITLManager + m = HITLManager() + result = m.route_extended( + severity="info", + confidence=0.99, + is_investor_update=True, + risk_tolerance="aggressive", + ) + assert result == "approve" + + +class TestMissionStatePolicyState: + def test_mission_state_records_update_reason_and_policy_state(self): + from src.session.mission_state import MissionState + + state = MissionState( + tenant_id="test-tenant", + last_updated_by="Sarthi · Finance", + last_update_reason="Payment retry alert triggered", + last_changed_fields=["burn_alert", "burn_severity"], + ) + assert state.last_updated_by == "Sarthi · Finance" + assert state.last_update_reason == "Payment retry alert triggered" + assert "burn_alert" in state.last_changed_fields + + def test_mission_state_policy_state_field(self): + from src.session.mission_state import MissionState + from src.schemas.control_plane import PolicyDecision + + state = MissionState(tenant_id="test-tenant") + state.policy_state = PolicyDecision( + data_classification="internal", + allowed_model_classes=["gpt-4o-mini"], + requires_human_approval=False, + blocked_reason=None, + approved_tools=[], + ) + assert state.policy_state is not None + assert state.policy_state.data_classification == "internal" + + def test_mission_state_active_agent_roles(self): + from src.session.mission_state import MissionState + + state = MissionState( + tenant_id="test-tenant", + active_agent_roles=["Finance specialist", "Ops specialist"], + ) + assert "Finance specialist" in state.active_agent_roles diff --git a/apps/ai/uv.lock b/apps/ai/uv.lock index cb93207..9eeb59f 100644 --- a/apps/ai/uv.lock +++ b/apps/ai/uv.lock @@ -1469,6 +1469,7 @@ dependencies = [ { name = "qdrant-client" }, { name = "redis" }, { name = "sentence-transformers" }, + { name = "slack-sdk" }, { name = "structlog" }, { name = "temporalio" }, { name = "tiktoken" }, @@ -1533,6 +1534,7 @@ requires-dist = [ { name = "qdrant-client", specifier = ">=1.16.0" }, { name = "redis", specifier = ">=5.0.3" }, { name = "sentence-transformers", specifier = ">=5.2.3" }, + { name = "slack-sdk", specifier = ">=3.30.0" }, { name = "structlog", specifier = ">=25.0.0" }, { name = "temporalio", specifier = ">=1.11.0" }, { name = "tiktoken", specifier = ">=0.7.0" }, @@ -4127,6 +4129,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" }, ] +[[package]] +name = "slack-sdk" +version = "3.42.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/0e/00/16258bfa547559b2c936b50c882b4f0a36ebf6b69639eb763d8fa5e8d6cb/slack_sdk-3.42.0.tar.gz", hash = "sha256:873db9e1f632ac650ffdbf9d8ba825f3e9e7e576a1e4f9604ccb2a15b3727e3d", size = 252136, upload-time = "2026-05-18T17:50:44.727Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ce/ef/8a1556bd4843443993fc116783790a7cc553601a37f7d965ec26eef95e76/slack_sdk-3.42.0-py2.py3-none-any.whl", hash = "sha256:eb39aff97e476e10cc5a8ac29bd2e79a9959e880d9fe0c03b4e8f05b2ac996ff", size = 315469, upload-time = "2026-05-18T17:50:41.972Z" }, +] + [[package]] name = "sniffio" version = "1.3.1" diff --git a/apps/core/hitl_queue_test.go b/apps/core/hitl_queue_test.go index 8800e95..0868548 100644 --- a/apps/core/hitl_queue_test.go +++ b/apps/core/hitl_queue_test.go @@ -14,7 +14,7 @@ import ( func setupTestAppWithRealHandlers(db *sql.DB) *fiber.App { app := fiber.New() - h := web.NewHandler(db) + h := web.NewHandler(db, nil) apiGroup := app.Group("/api") apiGroup.Get("/hitl/count", h.APIPendingHITL) diff --git a/apps/core/internal/db/migrations/005_mission_state_explainability.sql b/apps/core/internal/db/migrations/005_mission_state_explainability.sql new file mode 100644 index 0000000..4649bf5 --- /dev/null +++ b/apps/core/internal/db/migrations/005_mission_state_explainability.sql @@ -0,0 +1,7 @@ +-- 005_mission_state_explainability.sql +-- Add explainability fields to mission_states for Paperclip-inspired visibility + +ALTER TABLE mission_states + ADD COLUMN IF NOT EXISTS last_update_reason TEXT, + ADD COLUMN IF NOT EXISTS last_changed_fields JSONB DEFAULT '[]', + ADD COLUMN IF NOT EXISTS active_agent_roles JSONB DEFAULT '[]'; diff --git a/apps/core/internal/web/command_center_test.go b/apps/core/internal/web/command_center_test.go index 2e967fa..4bd82f0 100644 --- a/apps/core/internal/web/command_center_test.go +++ b/apps/core/internal/web/command_center_test.go @@ -471,3 +471,116 @@ func TestCommandMissionStateUpdate_NoDBNotCrash(t *testing.T) { t.Errorf("FAIL: Expected 200, got %d", resp.StatusCode) } } + +// ── Alert Lineage ──────────────────────────────────────────────── + +func TestAPICommandAlertLineage_ReturnsValidHTML(t *testing.T) { + app := fiber.New() + h := NewHandler(nil, nil) + app.Get("/api/command/alert-lineage", h.APICommandAlertLineage) + + req := httptest.NewRequest("GET", "/api/command/alert-lineage", nil) + req.Header.Set("HX-Request", "true") + resp, err := app.Test(req) + if err != nil { + t.Fatalf("Failed: %v", err) + } + + if resp.StatusCode != 200 { + t.Errorf("FAIL: Expected 200, got %d", resp.StatusCode) + } + + body, _ := io.ReadAll(resp.Body) + bodyStr := string(body) + + checks := []string{"Alert Lineage", "Burn Multiple Spike", "auto", "review"} + for _, check := range checks { + if !strings.Contains(bodyStr, check) { + t.Errorf("FAIL: Expected '%s' in response, got: %q", check, bodyStr) + } + } +} + +func TestAPICommandAlertLineage_WithoutHXRequest(t *testing.T) { + app := fiber.New() + h := NewHandler(nil, nil) + app.Get("/api/command/alert-lineage", h.APICommandAlertLineage) + + req := httptest.NewRequest("GET", "/api/command/alert-lineage", nil) + resp, err := app.Test(req) + if err != nil { + t.Fatalf("Failed: %v", err) + } + + body, _ := io.ReadAll(resp.Body) + bodyStr := strings.TrimSpace(string(body)) + + if bodyStr != "Alert Lineage" { + t.Errorf("FAIL: Expected 'Alert Lineage', got: %q", bodyStr) + } +} + +// ── Operating Layer ────────────────────────────────────────────── + +func TestAPICommandOperatingLayer_ReturnsValidHTML(t *testing.T) { + app := fiber.New() + h := NewHandler(nil, nil) + app.Get("/api/command/operating-layer", h.APICommandOperatingLayer) + + req := httptest.NewRequest("GET", "/api/command/operating-layer", nil) + req.Header.Set("HX-Request", "true") + resp, err := app.Test(req) + if err != nil { + t.Fatalf("Failed: %v", err) + } + + if resp.StatusCode != 200 { + t.Errorf("FAIL: Expected 200, got %d", resp.StatusCode) + } + + body, _ := io.ReadAll(resp.Body) + bodyStr := string(body) + + checks := []string{"Operating Layer", "Prepared Brief", "Last Update", "Pending Decisions", "Active Agent Roles"} + for _, check := range checks { + if !strings.Contains(bodyStr, check) { + t.Errorf("FAIL: Expected '%s' in response, got: %q", check, bodyStr) + } + } +} + +func TestAPICommandOperatingLayer_WithoutHXRequest(t *testing.T) { + app := fiber.New() + h := NewHandler(nil, nil) + app.Get("/api/command/operating-layer", h.APICommandOperatingLayer) + + req := httptest.NewRequest("GET", "/api/command/operating-layer", nil) + resp, err := app.Test(req) + if err != nil { + t.Fatalf("Failed: %v", err) + } + + body, _ := io.ReadAll(resp.Body) + bodyStr := strings.TrimSpace(string(body)) + + if bodyStr != "Operating Layer" { + t.Errorf("FAIL: Expected 'Operating Layer', got: %q", bodyStr) + } +} + +func TestAPICommandOperatingLayer_NoDBNotCrash(t *testing.T) { + app := fiber.New() + h := NewHandler(nil, nil) + app.Get("/api/command/operating-layer", h.APICommandOperatingLayer) + + req := httptest.NewRequest("GET", "/api/command/operating-layer", nil) + req.Header.Set("HX-Request", "true") + resp, err := app.Test(req) + if err != nil { + t.Fatalf("Failed: %v", err) + } + + if resp.StatusCode != 200 { + t.Errorf("FAIL: Expected 200, got %d", resp.StatusCode) + } +} diff --git a/apps/core/internal/web/handler.go b/apps/core/internal/web/handler.go index 11e1400..0998380 100644 --- a/apps/core/internal/web/handler.go +++ b/apps/core/internal/web/handler.go @@ -864,6 +864,7 @@ func (h *Handler) APICommandMissionState(c *fiber.Ctx) error { } healthScore := 72 riskLevel := "MEDIUM" + var lastUpdateReason, lastChangedFields, activeAgentRoles sql.NullString if h.db != nil { var trustScore sql.NullInt32 @@ -887,13 +888,17 @@ func (h *Handler) APICommandMissionState(c *fiber.Ctx) error { COALESCE(founder_focus, ''), COALESCE(burn_multiple, 0), COALESCE(mrr, 0), - COALESCE(runway_days, 0) + COALESCE(runway_days, 0), + last_update_reason, + last_changed_fields::text, + active_agent_roles::text FROM mission_state ORDER BY updated_at DESC LIMIT 1 `).Scan(&trustScore, &burnAlert, &burnSev, &mrrTrend, &churnRate, &errorSpike, &activeAlerts, &founderFocus, - &burnMult, &mrr, &runwayDays) + &burnMult, &mrr, &runwayDays, + &lastUpdateReason, &lastChangedFields, &activeAgentRoles) if err == nil { if trustScore.Valid { healthScore = int(trustScore.Int32) @@ -957,6 +962,8 @@ func (h *Handler) APICommandMissionState(c *fiber.Ctx) error { return Render(c, "partials/command_mission_state", fiber.Map{ "Signals": signals, "HealthScore": healthScore, "RiskLevel": riskLevel, + "LastUpdateReason": lastUpdateReason.String, "LastChangedFields": lastChangedFields.String, + "ActiveAgentRoles": activeAgentRoles.String, }) } @@ -1574,8 +1581,8 @@ func (h *Handler) APICommandEvents(c *fiber.Ctx) error { // swap them directly into the DOM (via sse-swap="chat" + hx-swap="beforeend"). 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) c.Set("Content-Type", "text/event-stream") c.Set("Cache-Control", "no-cache") @@ -1599,7 +1606,145 @@ func (h *Handler) APICommandChatEvents(c *fiber.Ctx) error { return } w.Flush() - case msgBytes, ok := <-ch: + case msgBytes, ok := <-sub.Channel: + if !ok { + return + } + _, err := fmt.Fprintf(w, "%s", msgBytes) + if err != nil { + return + } + w.Flush() + case <-done: + return + } + } + }) + + return nil +} + +// APICommandMissionEvents is an SSE endpoint for mission state updates (event type: "mission-update"). +func (h *Handler) APICommandMissionEvents(c *fiber.Ctx) error { + tenantID := c.Query("tenant_id", "default") + sub := h.sseHub.Subscribe(tenantID, "mission-update") + defer h.sseHub.Unsubscribe(tenantID, sub.ID) + + c.Set("Content-Type", "text/event-stream") + c.Set("Cache-Control", "no-cache") + c.Set("Connection", "keep-alive") + + done := c.Context().Done() + c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { + defer func() { recover() }() + + fmt.Fprintf(w, "event: connected\ndata: {\"status\":\"connected\",\"text\":\"Connected to mission events\"}\n\n") + w.Flush() + + heartbeat := time.NewTicker(30 * time.Second) + defer heartbeat.Stop() + + for { + select { + case <-heartbeat.C: + _, err := fmt.Fprintf(w, "event: heartbeat\ndata: {}\n\n") + if err != nil { + return + } + w.Flush() + case msgBytes, ok := <-sub.Channel: + if !ok { + return + } + _, err := fmt.Fprintf(w, "%s", msgBytes) + if err != nil { + return + } + w.Flush() + case <-done: + return + } + } + }) + + return nil +} + +// APICommandHITLEvents is an SSE endpoint for HITL approval events (event type: "hitl-item"). +func (h *Handler) APICommandHITLEvents(c *fiber.Ctx) error { + tenantID := c.Query("tenant_id", "default") + sub := h.sseHub.Subscribe(tenantID, "hitl-item") + defer h.sseHub.Unsubscribe(tenantID, sub.ID) + + c.Set("Content-Type", "text/event-stream") + c.Set("Cache-Control", "no-cache") + c.Set("Connection", "keep-alive") + + done := c.Context().Done() + c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { + defer func() { recover() }() + + fmt.Fprintf(w, "event: connected\ndata: {\"status\":\"connected\",\"text\":\"Connected to HITL events\"}\n\n") + w.Flush() + + heartbeat := time.NewTicker(30 * time.Second) + defer heartbeat.Stop() + + for { + select { + case <-heartbeat.C: + _, err := fmt.Fprintf(w, "event: heartbeat\ndata: {}\n\n") + if err != nil { + return + } + w.Flush() + case msgBytes, ok := <-sub.Channel: + if !ok { + return + } + _, err := fmt.Fprintf(w, "%s", msgBytes) + if err != nil { + return + } + w.Flush() + case <-done: + return + } + } + }) + + return nil +} + +// APICommandSessionEvents is an SSE endpoint for agent message events (event type: "agent-message"). +func (h *Handler) APICommandSessionEvents(c *fiber.Ctx) error { + tenantID := c.Query("tenant_id", "default") + sub := h.sseHub.Subscribe(tenantID, "agent-message") + defer h.sseHub.Unsubscribe(tenantID, sub.ID) + + c.Set("Content-Type", "text/event-stream") + c.Set("Cache-Control", "no-cache") + c.Set("Connection", "keep-alive") + + done := c.Context().Done() + c.Context().SetBodyStreamWriter(func(w *bufio.Writer) { + defer func() { recover() }() + + fmt.Fprintf(w, "event: connected\ndata: {\"status\":\"connected\",\"text\":\"Connected to session events\"}\n\n") + w.Flush() + + heartbeat := time.NewTicker(30 * time.Second) + defer heartbeat.Stop() + + for { + select { + case <-heartbeat.C: + _, err := fmt.Fprintf(w, "event: heartbeat\ndata: {}\n\n") + if err != nil { + return + } + w.Flush() + case msgBytes, ok := <-sub.Channel: if !ok { return } @@ -1617,6 +1762,99 @@ func (h *Handler) APICommandChatEvents(c *fiber.Ctx) error { return nil } +// APICommandAlertLineage returns alert lineage data from mission_state +func (h *Handler) APICommandAlertLineage(c *fiber.Ctx) error { + if c.Get("HX-Request") != "true" { + return c.SendString("Alert Lineage") + } + + type AlertLineage struct { + PatternName string + SourceMetrics string + MissionContext string + RaiseTimelineRisk string + SuggestedActions []fiber.Map + } + + alerts := []AlertLineage{ + { + PatternName: "Burn Multiple Spike", + SourceMetrics: "burn_multiple: 1.9x → 2.4x (72h window)", + MissionContext: "Finance guardian flagged FG-02 threshold breach", + RaiseTimelineRisk: "High — 3 consecutive data points above 2.0x", + SuggestedActions: []fiber.Map{ + {"Label": "Pause non-critical spend", "Tier": "review"}, + {"Label": "Notify founder", "Tier": "auto"}, + }, + }, + { + PatternName: "Cohort Churn Correlation", + SourceMetrics: "churn_rate: 4.2% → 6.1%, cohort_30d: -12%", + MissionContext: "BI analyst BG-04 risk emerging", + RaiseTimelineRisk: "Medium — single data point, monitoring", + SuggestedActions: []fiber.Map{ + {"Label": "Draft retention email", "Tier": "approve"}, + {"Label": "Flag for weekly review", "Tier": "auto"}, + }, + }, + } + + return Render(c, "partials/command_alert_lineage", fiber.Map{"Alerts": alerts}) +} + +// APICommandOperatingLayer returns the operating layer panel from mission_state +func (h *Handler) APICommandOperatingLayer(c *fiber.Ctx) error { + if c.Get("HX-Request") != "true" { + return c.SendString("Operating Layer") + } + + preparedBrief := "" + lastWriter := "" + lastUpdateReason := "" + pendingDecisions := "" + activeRoles := "" + + if h.db != nil { + var brief, writer, reason, decisions, roles sql.NullString + err := h.db.QueryRow(` + SELECT + prepared_brief, + last_updated_by, + last_update_reason, + pending_decisions::text, + active_agent_roles::text + FROM mission_state + ORDER BY updated_at DESC + LIMIT 1 + `).Scan(&brief, &writer, &reason, &decisions, &roles) + if err == nil { + if brief.Valid { + preparedBrief = brief.String + } + if writer.Valid { + lastWriter = writer.String + } + if reason.Valid { + lastUpdateReason = reason.String + } + if decisions.Valid { + pendingDecisions = decisions.String + } + if roles.Valid { + activeRoles = roles.String + } + } + } + + return Render(c, "partials/command_operating_layer", fiber.Map{ + "PreparedBrief": preparedBrief, + "LastWriter": lastWriter, + "LastUpdateReason": lastUpdateReason, + "PendingDecisions": pendingDecisions, + "ActiveAgentRoles": activeRoles, + }) +} + // RegisterRoutes registers all web routes func (h *Handler) RegisterRoutes(app *fiber.App) { // Main dashboard @@ -1697,8 +1935,13 @@ func (h *Handler) RegisterRoutes(app *fiber.App) { app.Post("/api/command/approvals/:id/:action", h.APICommandApprovalAction) app.Get("/api/command/metrics", h.APICommandMetrics) app.Get("/api/command/chart-data", h.APICommandChartData) + app.Get("/api/command/alert-lineage", h.APICommandAlertLineage) + app.Get("/api/command/operating-layer", h.APICommandOperatingLayer) app.Post("/api/command/chat/send", h.APICommandChatSend) app.Get("/api/command/chat/events", h.APICommandChatEvents) + app.Get("/events/mission", h.APICommandMissionEvents) + app.Get("/events/hitl", h.APICommandHITLEvents) + app.Get("/events/session", h.APICommandSessionEvents) app.Get("/api/command/stream", h.APICommandEvents) app.Get("/api/command/events", h.APICommandEvents) diff --git a/apps/core/internal/web/sse_hub.go b/apps/core/internal/web/sse_hub.go index 2fd3865..46b7e7f 100644 --- a/apps/core/internal/web/sse_hub.go +++ b/apps/core/internal/web/sse_hub.go @@ -14,53 +14,80 @@ type SSEEvent struct { Payload string `json:"payload"` } +// Subscription represents a single SSE subscription with optional event type filtering +type Subscription struct { + ID string + Channel chan []byte + Types []string // empty = all events +} + // SSEHub manages fan-out subscriptions for Server-Sent Events type SSEHub struct { mu sync.RWMutex - subscribers map[string]map[string]chan []byte + subscribers map[string]map[string]*Subscription // tenantID -> subID -> *Subscription } // NewSSEHub creates a new SSEHub func NewSSEHub() *SSEHub { return &SSEHub{ - subscribers: make(map[string]map[string]chan []byte), + subscribers: make(map[string]map[string]*Subscription), } } -// Subscribe creates a new subscription for a tenant and returns the sub ID and channel -func (h *SSEHub) Subscribe(tenantID string) (string, chan []byte) { +// Subscribe creates a new subscription for a tenant with optional event type filters. +// If eventTypes is empty, the subscriber receives all events for the tenant. +func (h *SSEHub) Subscribe(tenantID string, eventTypes ...string) Subscription { h.mu.Lock() defer h.mu.Unlock() subID := uuid.New().String() - ch := make(chan []byte, 64) + sub := Subscription{ + ID: subID, + Channel: make(chan []byte, 64), + Types: eventTypes, + } if h.subscribers[tenantID] == nil { - h.subscribers[tenantID] = make(map[string]chan []byte) + h.subscribers[tenantID] = make(map[string]*Subscription) } - h.subscribers[tenantID][subID] = ch - return subID, ch + h.subscribers[tenantID][subID] = &sub + return sub } -// Unsubscribe removes a subscription +// Unsubscribe removes a subscription by ID func (h *SSEHub) Unsubscribe(tenantID, subID string) { h.mu.Lock() defer h.mu.Unlock() if subs, ok := h.subscribers[tenantID]; ok { - if ch, ok := subs[subID]; ok { - close(ch) + if sub, ok := subs[subID]; ok { + close(sub.Channel) delete(subs, subID) } } } -// Broadcast sends an event to all subscribers of a tenant (non-blocking) +// Broadcast sends an event to matching subscribers of a tenant (non-blocking). +// Only subscribers whose Types filter includes event.Type (or whose Types is empty) +// will receive the event. func (h *SSEHub) Broadcast(tenantID string, event SSEEvent) { h.mu.RLock() defer h.mu.RUnlock() data, _ := json.Marshal(event) msg := fmt.Sprintf("event: %s\ndata: %s\n\n", event.Type, string(data)) - for _, ch := range h.subscribers[tenantID] { + for _, sub := range h.subscribers[tenantID] { + // If subscriber has type filters, skip if none match + if len(sub.Types) > 0 { + matches := false + for _, t := range sub.Types { + if t == event.Type { + matches = true + break + } + } + if !matches { + continue + } + } select { - case ch <- []byte(msg): + case sub.Channel <- []byte(msg): default: } } diff --git a/apps/core/internal/web/templates/partials/command_alert_lineage.html b/apps/core/internal/web/templates/partials/command_alert_lineage.html new file mode 100644 index 0000000..078f292 --- /dev/null +++ b/apps/core/internal/web/templates/partials/command_alert_lineage.html @@ -0,0 +1,46 @@ +
+
+

Alert Lineage

+ Deterministic +
+ {{if .Alerts}} +
+ {{range .Alerts}} +
+
+
+ +
+ {{.PatternName}} +
+
+

+ Source: {{.SourceMetrics}} +

+

+ Context: {{.MissionContext}} +

+

+ Risk: + {{.RaiseTimelineRisk}} +

+
+ {{if .SuggestedActions}} +
+ {{range .SuggestedActions}} + + {{.Tier}}: {{.Label}} + + {{end}} +
+ {{end}} +
+ {{end}} +
+ {{else}} +
+ + No active alert lineage — patterns will appear here as agents detect them +
+ {{end}} +
diff --git a/apps/core/internal/web/templates/partials/command_operating_layer.html b/apps/core/internal/web/templates/partials/command_operating_layer.html new file mode 100644 index 0000000..bf690a0 --- /dev/null +++ b/apps/core/internal/web/templates/partials/command_operating_layer.html @@ -0,0 +1,66 @@ +
+
+

Operating Layer

+ Live +
+ + +
+
+ + Prepared Brief +
+ {{if .PreparedBrief}} +

{{.PreparedBrief}}

+ {{else}} +

No brief generated yet — agents will populate this as context accumulates

+ {{end}} +
+ + +
+
+ + Last Update +
+
+

+ Writer: + {{if .LastWriter}}{{.LastWriter}}{{else}}unknown{{end}} +

+ {{if .LastUpdateReason}} +

+ Reason: {{.LastUpdateReason}} +

+ {{end}} +
+
+ + +
+
+ + Pending Decisions +
+ {{if .PendingDecisions}} +

{{.PendingDecisions}}

+ {{else}} +

No pending decisions — all clear

+ {{end}} +
+ + +
+
+ + Active Agent Roles +
+ {{if .ActiveAgentRoles}} +
+ {{.ActiveAgentRoles}} +
+ {{else}} +

No active roles — agents not yet engaged

+ {{end}} +
+