agent-chief 注意力守门员

72def stage1(
73    event,
74    *,
75    now: datetime,
76    policy: Policy,
77    quiet_hours: str = "23:00-08:00",
78    night_whitelist: list[str] | None = None,
79    recent_dedup_keys: frozenset[str] | set[str] = frozenset(),
80    embedder: Embedder = DEFAULT_EMBEDDER,
81) -> RuleHit | None:
82    """Stage-1 hard rules (SPEC §4.4). Returns a forced route, or None to continue."""
83    night_whitelist = night_whitelist or []
84
85    # Drop rules run before the quiet-hours digest rule: noise must die even at
86    # night, or it would resurface in the morning digest (see §4.7 events 1/24).
87    if policy.is_muted(event.topic):
88        return RuleHit("drop", "muted_topic", f"topic {event.topic} muted in POLICY.md")
89
90    if event.dedup_key and event.dedup_key in recent_dedup_keys:
91        return RuleHit("drop", "dedup", "duplicate of an event seen in the last 24h")
92
93    if is_zero_information(event.summary, embedder):
94        return RuleHit("drop", "zero_information", "zero-information template (empty report)")
95
96    if in_quiet_hours(now, quiet_hours) and not _topic_in_whitelist(
97        event.topic, night_whitelist
98    ):
99        return RuleHit("digest", "quiet_hours", f"quiet hours {quiet_hours}, topic not whitelisted")
100
101    rule = policy.route_for(event.topic)
102    if rule:
103        return RuleHit(
104            rule.route, f"policy:{rule.pattern}", f"POLICY.md rule {rule.pattern} -> {rule.route}"
105        )
106
107    return None

123class SimilarityClassifier:
124    """Compares an event against engaged/dismissed historical vectors (ms).
125
126    dismissed-sim > 0.88 with no engaged record → drop;
127    engaged-sim > 0.88 → skip judge, route by historical same-class mean;
128    otherwise → pass through to stage 3.
129    """
130
131    def __init__(self, embedder: Embedder = DEFAULT_EMBEDDER, threshold: float = STAGE2_THRESHOLD):
132        self.embedder = embedder
133        self.threshold = threshold
134        self._engaged: list[tuple[list[float], str]] = []  # (vector, route)
135        self._dismissed: list[list[float]] = []
136
137    def add_engaged(self, text: str, route: str) -> None:
138        self._engaged.append((self.embedder.embed(text), route))
139
140    def add_dismissed(self, text: str) -> None:
141        self._dismissed.append(self.embedder.embed(text))
142
143    def classify(self, text: str) -> Stage2Verdict:
144        vec = self.embedder.embed(text)
145        engaged_hits = [
146            (cosine(vec, evec), route)
147            for evec, route in self._engaged
148            if cosine(vec, evec) > self.threshold
149        ]
150        if engaged_hits:
151            # "route by historical same-class mean": majority route among the
152            # similar engaged records, nearest-first on ties.
153            engaged_hits.sort(reverse=True)
154            routes = [route for _, route in engaged_hits]
155            best = max(set(routes), key=lambda r: (routes.count(r), -routes.index(r)))
156            sim = engaged_hits[0][0]
157            return Stage2Verdict(
158                "route", best, sim, f"engaged-similar {sim:.2f}, historical route {best}"
159            )
160        dismissed_sim = max((cosine(vec, dvec) for dvec in self._dismissed), default=0.0)
161        if dismissed_sim > self.threshold:
162            return Stage2Verdict(
163                "drop", None, dismissed_sim,
164                f"dismissed-similar {dismissed_sim:.2f} with no engaged record",
165            )
166        return Stage2Verdict("pass", None, dismissed_sim, "unfamiliar; ask the judge")

115    async def process(self, payload: dict | Event) -> Decision:
116        from ingest.normalize import normalize
117
118        now = self.now_fn()
119        clock = _StageClock()
120        if isinstance(payload, Event):
121            event = payload
122        else:
123            event = await normalize(payload, inferrer=self.inferrer, now=now)
124        clock.mark("normalize")
125
126        policy = load_policy(self.policy_path)
127        scene = self._scene(now)
128        recent_keys = await self.state.recent_dedup_keys(now - timedelta(hours=24))
129
130        # Triage merge (SPEC §4.2 step 1): a same-topic near-duplicate within
131        # 10 min folds into the earlier event; the earlier decision stands.
132        if self.embedder:
133            recent = await self.state.recent_events(now - timedelta(minutes=10))
134            prior = find_mergeable(event, [e for e in recent if e.id != event.id],
135                                   embedder=self.embedder)
136            clock.mark("triage_merge", note="merged into prior" if prior else "no near-duplicate")
137            if prior:
138                merged = merge_events(prior, event)
139                await self.state.save_event(merged)
140                existing = await self.state.load_decision(prior.id)
141                if existing:
142                    # the prior decision (and its trace) stands for the merged event
143                    return existing.model_copy(
144                        update={"reason": existing.reason + "; merged near-duplicate"}
145                    )
146
147        hit = stage1(
148            event,
149            now=now,
150            policy=policy,
151            quiet_hours=self.quiet_hours,
152            night_whitelist=self.night_whitelist,
153            recent_dedup_keys=recent_keys,
154        )
155        if hit:
156            clock.mark("stage1", note=f"rule {hit.rule} fired")
157            decision = Decision(
158                event_id=event.id,
159                route=hit.route,  # type: ignore[arg-type]
160                scene=scene.scene,
161                scene_confidence=scene.confidence,
162                cost=0.0,
163                matched_rules=[hit.rule],
164                reason=hit.reason,
165                stage=1,
166                trace=DecisionTrace(stages=clock.stages),
167            )
168        else:
169            clock.mark("stage1", note="no hard rule fired")
170            decision = await self._stage2_or_judge(event, scene, policy, now, clock)

247        try:
248            result = await asyncio.wait_for(
249                self.judge.judge(event, context), self.judge_timeout
250            )
251        except Exception as exc:  # malformed output, timeout, backend down — never crash
252            logger.warning("judge %s unavailable (%s); degrading to rules-only", backend, exc)
253            last_error = f"{type(exc).__name__}: {exc}"[:200]
254            if self._degraded is not True or last_error != self._last_error:
255                # transition or changed failure mode: refresh the marker, but
256                # preserve `since` from any still-active record (incl. one
257                # written before a daemon restart mid-outage)
258                current = await load_degraded(self.state)
259                await self.state.set_meta(
260                    DEGRADED_KEY,
261                    {
262                        "active": True,
263                        "since": current["since"] if current else now.isoformat(),
264                        "last_error": last_error,
265                    },
266                )
267                self._degraded, self._last_error = True, last_error
268            clock.mark("judge", note=f"FAILED: {type(exc).__name__}")
269            # retries that failed were still paid for — bill them (JudgeError.usage)
270            usage = getattr(exc, "usage", None)
271            return Decision(
272                event_id=event.id,
273                route="digest",  # conservative: never interrupt, never drop, while blind
274                scene=scene.scene,
275                scene_confidence=scene.confidence,
276                cost=self._bill(backend, usage),
277                reason=f"judge unavailable ({type(exc).__name__}); "
278                "conservative rules-only routing to digest",
279                stage=3,
280                degraded=True,
281                trace=self._usage_trace(clock, backend, usage),
282            )

207def score_and_route(
208    result,
209    scene: SceneState,
210    *,
211    topic_weights: dict[str, float] | None = None,
212    scene_cost: float = 0.0,
213    threshold_overrides: dict[str, float] | None = None,
214    threshold_adjust: float = 0.0,
215    memory_hit: bool = False,
216) -> tuple[str, float, dict[str, float], str]:
217    """Compose `score = Σ(w_topic[dim]·comp[dim]) − scene_cost` and route.
218
219    Returns (route, score, components-after-boost, reason).
220    """
221    from context.infer import downgrade_low_confidence, interrupt_threshold
222
223    weights = {**DEFAULT_WEIGHTS, **(topic_weights or {})}
224    comps = {dim: getattr(result, dim) for dim in DIMS}
225    if memory_hit:
226        comps["relevance"] = min(1.0, comps["relevance"] * 1.2)
227
228    score = sum(weights[dim] * comps[dim] for dim in DIMS) - scene_cost
229    threshold = interrupt_threshold(scene.scene, threshold_overrides)
230    if threshold_adjust:
231        # learned global adjustment, clamped (SPEC §4.6 bounds [0.35, 0.95])
232        threshold = min(0.95, max(0.35, threshold + threshold_adjust))
233
234    if score >= threshold:
235        route = "interrupt"
236        reason = f"score {score:.2f} ≥ {scene.scene} threshold {threshold:.2f}"
237    elif score >= DIGEST_FLOOR:
238        route = "digest"
239        reason = f"score {score:.2f} below {scene.scene} threshold {threshold:.2f}"
240    elif result.memorize:
241        route = "curate"
242        reason = f"score {score:.2f} low but worth remembering"
243    else:
244        route = "drop"
245        reason = f"score {score:.2f} with no lasting value"
246
247    route = downgrade_low_confidence(route, scene)
248    if route == "digest" and score >= threshold:
249        reason += f"; downgraded (scene confidence {scene.confidence:.2f} < 0.6)"
250
251    # SPEC §4.4: dispatchable prep work runs first, then delivery ("arrive with a plan")
252    if result.dispatchable and route in ("interrupt", "digest"):
253        route = "dispatch"
254        reason += "; dispatchable prep work available"
255
256    return route, score, comps, reason

19class Event(BaseModel):
20    id: str
21    source: str
22    topic: str
23    summary: str = Field(max_length=200)
24    detail: str | None = None
25    suggested_action: str | None = None
26    evidence: list[str] = []
27    claimed_urgency: Literal["low", "medium", "high"] | None = None
28    expires_at: datetime | None = None
29    dedup_key: str | None = None
30    received_at: datetime
31
32
33class StageTiming(BaseModel):
34    """One pipeline stage in a decision trace (SPEC v3.1 Step 26)."""
35
36    stage: str  # triage_merge | stage1 | stage2 | associate | judge | route
37    ms: float
38    note: str = ""
39
40
41class DecisionTrace(BaseModel):
42    """Per-decision latency, token and cost accounting (SPEC v3.1 Step 26)."""
43
44    stages: list[StageTiming] = []
45    tokens_in: int = 0
46    tokens_out: int = 0
47    cached_tokens: int = 0
48    backend: str | None = None
49    prompt_version: str | None = None
50
51
52class Decision(BaseModel):
53    event_id: str
54    route: Route
55    score: float | None = None
56    components: dict[str, float] | None = None
57    scene: str
58    scene_confidence: float
59    cost: float  # USD cost of this judgment (0.0 when no LLM was consulted)
60    matched_rules: list[str] = []
61    reason: str
62    stage: int
63    dispatch_task_id: str | None = None
64    trace: DecisionTrace | None = None
65    degraded: bool = False  # judge unavailable → rules-only conservative routing (Step 28)

15class MemoryStore:
16    def __init__(self, state: State, embedder: Embedder = DEFAULT_EMBEDDER):
17        self.state = state
18        self.embedder = embedder
19
20    async def curate(
21        self, text: str, *, topic: str, origin_event_id: str | None, now: datetime
22    ) -> MemoryItem:
23        item = MemoryItem(
24            id=f"mem_{now:%Y%m%d}_{secrets.token_hex(3)}",
25            origin_event_id=origin_event_id,
26            text=text,
27            topic=topic,
28            embedding=self.embedder.embed(text),
29            created_at=now,
30        )
31        await self.state.save_memory(item)
32        return item
33
34    async def associate(self, text: str, *, now: datetime) -> list[MemoryItem]:
35        """Top-3 active memories with cosine > 0.78; updates hit stats (SPEC §4.2)."""
36        query = self.embedder.embed(text)
37        scored = [
38            (cosine(query, item.embedding), item)
39            for item in await self.state.list_memory()
40        ]
41        hits = sorted(
42            (pair for pair in scored if pair[0] > ASSOCIATION_THRESHOLD),
43            key=lambda pair: pair[0],
44            reverse=True,
45        )[:TOP_K]
46        results = []
47        for _sim, item in hits:
48            item.hit_count += 1
49            item.last_hit_at = now
50            await self.state.save_memory(item)
51            results.append(item)
52        return results
53
54    async def expire(self, *, now: datetime) -> int:
55        """Move items past their ttl_days into the archive table (SPEC §3)."""
56        moved = 0
57        for item in await self.state.list_memory():
58            if now - item.created_at > timedelta(days=item.ttl_days):
59                await self.state.save_memory(item, archive=True)
60                await self.state.delete_memory(item.id)
61                moved += 1
62        return moved

30async def verify(
31    task: Task, *, ask: AskFn | None = None, exec_fn: Callable = _default_exec
32) -> VerifyResult:
33    """Check a completed task against its acceptance criteria."""
34    if task.acceptance_cmd:
35        argv = shlex.split(task.acceptance_cmd)
36        code, _out, err = await exec_fn(argv, None)
37        if code == 0:
38            return VerifyResult(True, "acceptance_cmd exit 0")
39        return VerifyResult(False, f"acceptance_cmd exit {code}: {err.strip()[:120]}")
40
41    if ask is None:
42        # No verifier available: never bless unverified work.
43        return VerifyResult(False, "no verifier configured; refusing to trust the claim")
44
45    prompt = verify_prompt(
46        acceptance=task.acceptance, result=task.result_summary or "(no result)"
47    )
48    raw = await ask(prompt)
49    try:
50        data = json.loads(extract_json(raw))
51        return VerifyResult(data.get("verdict") == "pass", data.get("reason", ""))
52    except Exception:
53        logger.warning("verifier returned garbage; treating as fail: %r", raw[:80])
54        return VerifyResult(False, "verifier output unparseable")
55
56
57async def dispatch_and_verify(
58    state: State,
59    task: Task,
60    executor: Executor,
61    *,
62    ask: AskFn | None = None,
63    exec_fn: Callable = _default_exec,
64) -> tuple[Task, str | None]:
65    """Run + verify with one retry (attempts max 2, SPEC §3); on final failure
66    mark the task rejected and return a message asking the human."""
67    last_reason = ""
68    while task.attempts < 2:
69        task = await run_task(state, task, executor)
70        if task.status == "done":
71            result = await verify(task, ask=ask, exec_fn=exec_fn)
72            if result.passed:
73                return task, None
74            last_reason = result.reason
75            task.status = "failed"
76            await state.save_task(task)
77        else:
78            last_reason = task.result_summary or "executor failed"
79
80    task.status = "rejected"
81    await state.save_task(task)
82    ask_human = (
83        f"I tried twice and couldn't finish this myself: {task.goal}\n"
84        f"Last attempt: {task.result_summary}\n"
85        f"Why it didn't pass: {last_reason}\n"
86        f"Can you take a look?"
87    )
88    return task, ask_human