Brain loop 与降级恢复 深度解析

49class Brain:
50    """The triage→associate→decide loop for one incoming event (SPEC §4.2)."""
51
52    def __init__(
53        self,
54        state: State,
55        judge: Judge,
56        *,
57        policy_path,
58        quiet_hours: str = "23:00-08:00",
59        night_whitelist: list[str] | None = None,
60        scene_engine=None,
61        classifier: SimilarityClassifier | None = None,
62        memory=None,
63        inferrer=None,
64        shadow=None,
65        audit: AuditLog | None = None,
66        user_profile: str = "",
67        default_executor: str = "claude_code",
68        embedder=None,
69        actor: Callable | None = None,
70        now_fn: Callable[[], datetime] = lambda: datetime.now(UTC),
71        judge_timeout: float = 150.0,  # > HTTPJudge worst case (2 attempts x 60s) + slack
72    ):
73        self.state = state
74        self.judge = judge
75        self.policy_path = policy_path
76        self.quiet_hours = quiet_hours
77        self.night_whitelist = night_whitelist or ["family", "production_incident"]
78        self.scene_engine = scene_engine
79        self.classifier = classifier
80        self.memory = memory
81        self.inferrer = inferrer
82        self.shadow = shadow
83        self.audit = audit
84        self.user_profile = user_profile
85        self.default_executor = default_executor
86        self.embedder = embedder
87        self.actor = actor  # async (Event, Decision) -> None; delivery/dispatch side
88        self.now_fn = now_fn
89        self.judge_timeout = judge_timeout
90        self._degraded: bool | None = None  # None = unknown until first judgment
91        self._last_error: str | None = None

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)
171
172        if self.shadow:
173            route, annotation = await self.shadow.apply(decision, now=now)
174            if annotation:
175                decision = decision.model_copy(
176                    update={"route": route, "reason": f"{decision.reason}; {annotation}"}
177                )
178
179        await self.state.save_event(event)
180        await self.state.save_decision(decision)
181        if self.actor:
182            asyncio.ensure_future(self._act_safely(event, decision))
183        if self.audit:
184            self.audit.write(
185                {
186                    "at": now.isoformat(),
187                    "event_id": event.id,
188                    "route": decision.route,
189                    "stage": decision.stage,
190                    "score": decision.score,
191                    "scene": decision.scene,
192                    "reason": decision.reason,
193                    "cost": decision.cost,
194                    "prompt_version": decision.trace.prompt_version if decision.trace else None,
195                    "degraded": decision.degraded,
196                }
197            )
198        return decision

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            )
283        if self._degraded is not False:  # recovery (or first success): clear marker once
284            await self.state.set_meta(DEGRADED_KEY, {"active": False})
285            self._degraded = False
286        clock.mark("judge", note=f"backend {backend}")