File size: 4,407 Bytes
0721bb4 0066161 0721bb4 0066161 0721bb4 0066161 0721bb4 0066161 0e02a0f 0721bb4 0e02a0f 0721bb4 0e02a0f 0721bb4 0e02a0f 0721bb4 0066161 0721bb4 0066161 0721bb4 0066161 0e02a0f 0721bb4 0066161 0721bb4 0066161 0721bb4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """Deterministic routing gate — policy check over the router's intent.
After the LLM router picks an intent, the gate checks it against the per-analysis
Analysis State and returns the **effective** intent: allow as-is, or redirect. No
LLM, no I/O in `gate()` itself.
Only one rule has teeth in v1: an analytical request (`structured_flow`) requires a
validated problem statement (`problem_validated is True`); otherwise it is
redirected to `problem_statement` so the user defines the goal first. Everything
else passes through. `generate_report` is not a router intent (button / report
API), so it is not gated here.
`AnalysisState` is the locked 8-field contract (mirrors the `analysis_states` DB
table). `get_analysis_state` reads the real per-analysis row via `AnalysisStateStore`
(#9, landed); it fails closed to a not-validated stub on a missing row or read error.
See `ORCHESTRATOR_REWORK_PLAN.md` §4.
"""
from __future__ import annotations
from datetime import UTC, datetime
from pydantic import BaseModel, Field
from src.agents.orchestration import Intent
from src.middlewares.logging import get_logger
logger = get_logger("gate")
class AnalysisState(BaseModel):
"""Per-analysis state the Help skill + report layer read every turn.
Field names mirror the dedorch `analyses` table so the DB read swaps in without
touching readers. The goal is the user-entered `objective` + `business_questions`
(set at onboarding by Go); the old `problem_statement`/`problem_validated` gate fields
were dropped (dedorch #3 / KM-652). `report_id` is null until a report exists.
"""
id: str
analysis_title: str
objective: str = ""
business_questions: list[str] = Field(default_factory=list)
user_id: str
report_id: str | None = None
created_at: datetime
updated_at: datetime
def gate(intent: Intent, state: AnalysisState) -> Intent:
"""Return the effective intent (NEUTERED 2026-06-24 — passes everything through).
The `problem_validated` gate was removed: analysis is no longer gated on a validated
problem statement (the goal is now two user-entered fields, `objective` +
`business_questions`, captured at onboarding with no agent validation). Kept as a
no-op seam so gating can be restored without re-threading call sites.
"""
# Pre-2026-06-24 policy: redirect analytical requests until the goal was validated.
# if intent == "structured_flow" and not state.problem_validated:
# logger.info(
# "gate redirect",
# requested=intent,
# effective="problem_statement",
# reason="problem_not_validated",
# )
# return "problem_statement"
return intent
def stub_analysis_state() -> AnalysisState:
"""Hardcoded Analysis State for never-throw fallbacks / tests.
Shared fixture so the gate seam, the Help skill, and tests all exercise the same
shape when a real row is missing or a read fails.
"""
now = datetime.now(UTC)
return AnalysisState(
id="stub-analysis",
analysis_title="Stub analysis",
objective="",
business_questions=[],
user_id="stub-user",
report_id=None,
created_at=now,
updated_at=now,
)
async def get_analysis_state(analysis_id: str) -> AnalysisState:
"""Load the Analysis State for an analysis (shared id with the chat room).
Reads the `analysis_states` row via `AnalysisStateStore`. Never-throw seam: a
missing row (e.g. a legacy room created before this table) or a read failure
degrades to a **not-validated** stub, so the gate fails closed (→ steer to
`problem_statement`) rather than running ungated analysis. The store import is
lazy so this module stays import-safe without a DB.
"""
try:
from src.agents.state_store import AnalysisStateStore
state = await AnalysisStateStore().get(analysis_id)
except Exception as exc: # noqa: BLE001 — never-throw; fail closed to not-validated
logger.warning(
"get_analysis_state read failed — default not-validated",
analysis_id=analysis_id,
error=str(exc),
)
return stub_analysis_state()
if state is None:
logger.debug("analysis_state missing — default not-validated", analysis_id=analysis_id)
return stub_analysis_state()
return state
|