diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md index bec898654e..114a7b7e75 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md @@ -110,7 +110,7 @@ P0 blocks correctness or continuity in the current user journey. P1 enables repe | **S4 Runtime/host/daemon · P0/P1** | Attached/managed, Turn, broker, runtime connectors and Desktop repairs exist; registration does not establish executable capacity | Qualify multi-Turn supervision for one real supported combination; restart/cancel/drain/stop retain work and fence old executors. Then expand host parity, unique service-profile ownership, clean installation and upgrades; show unsupported adapter capabilities | | **S5 Frontend, Lark and human interaction · P0/P1** | Local chat, settings, proposals and partial Goal Channel verticals exist; Goal Overview, Tasks, the W1 work map, usage, acceptance readback and lifecycle stop ship as separate regions without one page owner; shared audience/session/work readback needs qualification | One journey spans settings, work graph, handoff, blockers, cost, corrections, artifacts and return. **Page:** the [Goal workspace](goal-workspace-surface-v0.md) composes that journey per Goal over shared typed projections: a persistent masthead with scoped Stop, vitals with one meaning per figure (named window and coverage, unknown is not zero), work grouped by owner, results and an all-work rail. M1–M2 come first, with concrete first-screen evidence in the PR description; authorized refinements need no repeated confirmation. **Cases:** reconnect/repeated-click/stale/original-route. **Realtime IM** reuses Chat/Turn: isolate independent conversations on one listener, then qualify ordinary DM onboarding, explicit role selection, busy-session admission, provisional progress, media and exact permission callbacks through [the shared operational contract](capable-manager-semantic-handoff-v0.md#10-operational-contract). **Team:** [Live team workspace](live-team-workspace-v0.md) makes exchange, revision and original-coordinator continuation visible. Its [Work-scale map track](live-team-workspace-v0.md#11-delivery-order-and-relationship-to-aggressive-r2-progress) draws each Goal's typed Todo relations first (W1, the Goal workspace's Map lens), then live state and outputs on the same nodes. Then intelligent review, keyboard accessibility and a launcher-only command palette, bilingual terminology, actionable errors and offline degradation; interrupt only for actual decisions | | **S6 Materials, evidence, memory and learning · P1** | Authority registry, material lifecycle/frontier, decision context, reward memory and turn recall exist; explicit project material packets and source-verifier gates are implemented, while private source initialization/Core binding and ordinary-conversation adoption remain unqualified; direction baseline and parts of attribution remain proposed | Connect material revision→same-Agent read→decision reference→artifact/outcome. Expose expiry/revocation/source loss and forgetting policy. Handoff preserves decision-relevant summaries and authorized artifacts; qualify OpenViking/Obelisk as optional providers. Prove causal utility with controls, not relevance alone | -| **S7 Budget, scheduling and fleet scale · P0 observation/P1–P2 expansion** | Quota/scheduler and partial usage aggregates exist; full provider cost, distributed reservations and hundred-Agent concurrency need evidence | Separate configured budget, admission, consumption and estimates; unknown is not zero and replay cannot double-charge. R7 pagination/bounded summaries and [complete-history transport](typescript-control-plane-migration-v0.md), including refresh/replay/single-debit evidence beyond the RPC limit; provider/host limits, fairness, backpressure, event wake and isolation; report registration/activity/throughput and cost per accepted outcome separately. Receipt-backed semantic progress closes its original host Turn without completing an open/waiting Todo; File/SQLite guard replay preserves verified writeback when a later dependency wait changes the frontier and covers automatic quiet writeback during a receipt-owned retry, preserving independent work and monitor clocks. The typed full-source batch keeps receipt retry identity independently of the earliest wakeup, reconciles earlier retries with retained caches/hidden clocks, and keeps ordinary notices non-gating across earlier/equal/later monitors; File/SQLite replay and canonical readback qualify this source boundary. [Full quota and TurnEnvelope](../../reference/protocols/turn-envelope-v0.md) retain the original executable settlement command and single debit. Follow [the quota settlement contract](../../quota-allocation.md#receipt-backed-settlement-progress). External-wait authoring now derives successor readiness from the locked Todo snapshot through the existing resume owner, covering completed dependencies and monitor generation fences without trusting display flags. File/SQLite CLI coverage also rejects an original Turn’s step write into a newly selected Todo, preserves that Todo’s step/CAS and dependencies, and settles the original writeback once. Recommendation consumers reuse the v0/v1 settlement identity owner: Todo-less replan rejects Todo-step editing with actionable recovery, while derived or explicit guidance preserves its original once-only settlement and monitor state. Source-qualified continuation, installed adoption, live Host waiting-state cadence/readback and fleet scale remain separate acceptance; this bounded repair does not qualify fleet scale or host latency. | +| **S7 Budget, scheduling and fleet scale · P0 observation/P1–P2 expansion** | Quota/scheduler and partial usage aggregates exist; full provider cost, distributed reservations and hundred-Agent concurrency need evidence | Separate configured budget, admission, consumption and estimates; unknown is not zero and replay cannot double-charge. R7 pagination/bounded summaries and [complete-history transport](typescript-control-plane-migration-v0.md), including refresh/replay/single-debit evidence beyond the RPC limit; provider/host limits, fairness, backpressure, event wake and isolation; report registration/activity/throughput and cost per accepted outcome separately. Receipt-backed semantic progress closes its original host Turn without completing an open/waiting Todo; File/SQLite guard replay preserves verified writeback when a later dependency wait changes the frontier and covers automatic quiet writeback during a receipt-owned retry, preserving independent work and monitor clocks. The typed full-source batch keeps receipt retry identity independently of the earliest wakeup, reconciles earlier retries with retained caches/hidden clocks, and keeps ordinary notices non-gating across earlier/equal/later monitors; File/SQLite replay and canonical readback qualify this source boundary. [Full quota and TurnEnvelope](../../reference/protocols/turn-envelope-v0.md) retain the original executable settlement command and single debit. Follow [the quota settlement contract](../../quota-allocation.md#receipt-backed-settlement-progress). External-wait authoring now derives successor readiness from the locked Todo snapshot through the existing resume owner, covering completed dependencies and monitor generation fences without trusting display flags. File/SQLite CLI coverage also rejects an original Turn’s step write into a newly selected Todo, preserves that Todo’s step/CAS and dependencies, and settles the original writeback once. Recommendation consumers reuse the v0/v1 settlement identity owner: Todo-less replan rejects Todo-step editing with actionable recovery, while derived or explicit guidance preserves its original once-only settlement and monitor state. Source-qualified continuation, installed adoption, live Host waiting-state cadence/readback and fleet scale remain separate acceptance; this bounded repair does not qualify fleet scale or host latency. The bounded current-streak repair composes one typed Monitor expiry rule across status and Goal-frontier callers. File/SQLite Codex CLI readback preserves counters and pending work: expired/equal-time Monitors no longer cause a current-streak replan; active future-due and independent history triggers remain. Source behavior, installed adoption and full Goal/CI qualification stay separate. | | **S8 Capabilities, extensions and domain integration · P1/P2** | Capability catalog, extension lifecycle, hooks, engineering/research/content/office capabilities and computer-use contracts exist | First exercise the shared control plane with existing issue-fix/PR-review and material/research callers. Every provider has readiness/version/permissions/default-off/uninstall/rollback/isolation and real-entry evidence. New domain effects start with one simulated operation, not a marketplace or workflow DSL | | **S9 Identity, authority, privacy and trust · continuous P0/P1–P2 remote** | Public/private scope, capability gates, fencing and confirmation contracts belong to existing owners | R1/R3 cover sender/audience/artifact scope and stale authority; R6 authenticates tenant/Goal/actor/host, rotation/revocation and least privilege. Qualify credential custody, untrusted tool/document inputs, dependency supply chain, audit retention/deletion and vulnerability response through real paths; roles/messages/memory mint no write authority | | **S10 Reliability, diagnostics and operations · P0/P1** | Recovery/canary, read-only diagnostics prototype and DSH event adapter exist; C0/C1, overhead and full operations qualification are open | Failure classification→observable state→recovery drill→regression prevention; process/storage/network/delivery failures and data growth. Accepted Chat requests must settle even when runtime preparation fails before dispatch; qualify missing runtime assets, stop races and recovery without replay under [the shared conversation operational contract](capable-manager-semantic-handoff-v0.md#10-operational-contract). A context/provider read returning after the stop wait must honor the persisted Turn and exact Session claim: no late provider dispatch or handoff, even when a fresh request has completed. This bounded GQ08 repair does not certify upstream interrupt fidelity or effects already admitted elsewhere. Use [bounded repair lookup and targeted diagnostics](../../../skills/loopx-self-repair/references/targeted-diagnostics.md) to reduce redundant reads above the provider boundary; measure backend-specific cold/warm reads, writes and lock waits separately. Freeze SLO/RPO/RTO/capacity/retention boundaries and measure before qualification. Lark transport backpressure and drain retain the App consumer lease; unpersisted buffered lines are not accepted work or completion receipts. Runbooks include upgrade, restore, stop and human takeover; test counts do not prove recovery | diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md index 4e716605c0..094ac843f4 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md @@ -93,7 +93,7 @@ managed 与 attached 的工作对话都应能持续在 LoopX 中进行：沿用 | **S4 runtime/host/daemon · P0/P1** | attached/managed、Turn、broker、runtime connector 和 Desktop 修复存在；“registered”不等于可执行 | 选择一个真实合格组合完成多 Turn supervision；restart/cancel/drain/stop 不丢工作且旧 executor 被 fence。之后扩 host parity、service-profile 唯一 owner、干净安装与版本升级；按 adapter 能力显示不支持项 | | **S5 前端、Lark 与人机交互 · P0/P1** | 本地对话、settings、proposal 和部分 Goal Channel vertical 已有；Goal 概览、任务、W1 工作地图、用量、验收回读与生命周期停止作为分散区域已交付，但没有统一页面 owner；统一受众/会话/工作回读仍需资格 | 用一个团队旅程贯穿设置、工作图、handoff、阻塞、成本、修订、产物和回报。**页面：**[Goal 工作区](goal-workspace-surface-v0.zh-CN.md)在共享 typed projection 之上按 Goal 组合这条旅程：带限定范围停止的常驻页头、每个数字只有一种含义的指标（标明窗口与覆盖度，未知不是零）、按负责人分组的工作、成果与全部工作侧栏；先做 M1–M2，将具体首屏放入 PR description；获准方向内的细化不重复确认。**场景：**验证重连/重复点击/stale/原路反馈。**实时 IM** 复用 Chat/Turn：先让同一 listener 下的独立会话互不阻塞，再按[共享运行契约](capable-manager-semantic-handoff-v0.zh-CN.md#10-运行契约)验收普通私聊开通、显式角色选择、忙碌会话准入、临时进度、媒体和精确权限回调。**团队：**[团队实时工作区](live-team-workspace-v0.zh-CN.md)让交换、修订与原协调员继续推进可见；其[工作尺度地图分线](live-team-workspace-v0.zh-CN.md#11-交付顺序与激进推进-r2-的关系)先画出每个 Goal 的类型化 Todo 关系（W1，即 Goal 工作区的地图镜头），再在同一节点叠加实时状态与产出。再做 intelligent review、无障碍键盘流程与只作启动器的命令面板、中英术语、错误可恢复和离线降级；只在真实决策处打断人 | | **S6 材料、证据、记忆与学习 · P1** | authority registry、material lifecycle/frontier、decision context、reward memory、turn recall 已有；方向基线和部分归因仍是提案 | 先打通“材料 revision→同 Agent 阅读→决策引用→产物/结果”；失效、撤销、来源消失与遗忘策略可回读。handoff 保存影响决策的摘要与授权 artifact；OpenViking/Obelisk 按可选 provider 资格化。utility 的因果收益另以对照证明，不把相关性当提升 | -| **S7 预算、调度与 fleet 规模 · P0 观测/P1–P2 扩展** | quota/scheduler 与部分 usage aggregate 存在；全 provider 成本、分布式资源预留及百 Agent 并发尚需证据 | 先区分配置预算、准入、消耗与估算；未知成本不记零、重复事件不双记。R7 分页/有界摘要及[完整历史传输](typescript-control-plane-migration-v0.zh-CN.md)，验收超出 RPC 上限后的写回/重放/单次扣记；provider/host 限流、公平性、背压、事件唤醒与失败隔离；分别报告注册数/活跃数/吞吐量和每个验收成果成本 回执支持的语义进展已能关闭原 host Turn，保留未完成/等待中的 Todo；file/SQLite guard 回放覆盖 frontier 改变，边界见[quota 结算契约](../../quota-allocation.md#receipt-backed-settlement-progress)。external-wait authoring 现复用 resume owner 从锁定 Todo 快照重算后继，以完成依赖/monitor 代际证明恢复，不再相信展示字段；源证据续接与真实 Host backoff/readback 仍须分别验收。这项有界修复不证明 fleet 规模或宿主延迟。 | +| **S7 预算、调度与 fleet 规模 · P0 观测/P1–P2 扩展** | quota/scheduler 与部分 usage aggregate 存在；全 provider 成本、分布式资源预留及百 Agent 并发尚需证据 | 先区分配置预算、准入、消耗与估算；未知成本不记零、重复事件不双记。R7 分页/有界摘要及[完整历史传输](typescript-control-plane-migration-v0.zh-CN.md)，验收超出 RPC 上限后的写回/重放/单次扣记；provider/host 限流、公平性、背压、事件唤醒与失败隔离；分别报告注册数/活跃数/吞吐量和每个验收成果成本 回执支持的语义进展已能关闭原 host Turn，保留未完成/等待中的 Todo；file/SQLite guard 回放覆盖 frontier 改变，边界见[quota 结算契约](../../quota-allocation.md#receipt-backed-settlement-progress)。external-wait authoring 现复用 resume owner 从锁定 Todo 快照重算后继，以完成依赖/monitor 代际证明恢复，不再相信展示字段；源证据续接与真实 Host backoff/readback 仍须分别验收。这项有界修复不证明 fleet 规模或宿主延迟。 当前连续计数有界修复让 status 与 Goal-frontier 调用方共用一个 typed Monitor 过期规则。File/SQLite Codex CLI 回读保留计数及等待中的工作：过期及恰好到期的 Monitor 不再引起当前计数重规划；活跃但未来 due 的监控及独立历史触发保持。源码行为、安装采用与完整 Goal/CI 资格分别验收。 | | **S8 能力、扩展与领域集成 · P1/P2** | 已有 capability catalog、extension 生命周期、hook、工程/研究/content/office 能力及 computer-use 合同 | 优先用现有 issue-fix/PR-review 和材料/研究 caller 检验共享控制面；每个 provider 带 readiness、版本、权限、默认关闭、卸载/回滚、失败隔离与真实入口证据。新 domain effect 从模拟单操作闭环开始，不先建市场或通用工作流 DSL | | **S9 身份、权限、隐私与信任 · P0 持续/P1–P2 远端** | public/private 边界、作用域、capability gate、fence 与确认合同分布在已有 owner | 随 R1/R3 验 sender/audience/artifact scope 和 stale authority；远端 R6 必须认证 tenant/Goal/actor/host、轮换撤销与最小权限。凭据保管、非可信工具/文档输入、依赖供应链、审计留存/删除及漏洞响应纳入真实路径；角色、消息或 memory 不铸造写权限 | | **S10 可靠性、诊断与运行运营 · P0/P1** | recovery/canary、read-only diagnostics 原型及 DSH event adapter 已有；C0/C1、开销和完整运营资格仍未闭合 | 故障分类→可观察状态→恢复演练→防复发；覆盖进程/存储/网络/投递故障和数据增长。Chat 上下文或 provider 读取晚于停止等待返回时，按持久 Turn 和精确 Session claim 判断：即使新请求已完成，也不得再启动旧请求或交接迟到结果。这项有界 GQ08 修复不证明上游 interrupt 保真，也不取消其他 owner 已准入的效果；完整恢复仍遵循[共享对话运行契约](capable-manager-semantic-handoff-v0.md#10-operational-contract)。定义并冻结 SLO、RPO/RTO、容量/保留边界，实测后标 qualified；Lark 传输背压与 drain 保留 App consumer lease；未持久化的缓冲消息不算已受理工作或完成回执。运行手册含升级、备份恢复、停止与人工接管，不以测试数代替恢复结果 | diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index f6d780f734..45c7c0bd0d 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -984,6 +984,20 @@ receipts have no concrete executed-monitor identity and do not count toward this threshold; distinct heartbeat turn ids are idempotency evidence, not monitor execution evidence. +The current Todo `consecutive_no_change` detector shares the Monitor lifetime +rule with scheduling. At one explicit runtime observation, `expires_at` at or +before that instant excludes that Monitor from `monitor_no_change_streak`. +The recorded Todo, counters, observation hash/generation and pending successor +remain unchanged. An active Monitor whose next poll is in the future can still +require a streak review; expiry and due time have different meanings. +Goal-frontier and status callers use the same TypeScript expiry/selection owner +while preserving their existing trigger shape, scope and tie order. Historical +requests without a current clock retain their previous replay behavior; +independent progress and periodic-review triggers remain effective. Quiet +heartbeat liveness receipts can still be recorded without polling the source +or spending a slot. This source repair does not establish installed adoption, +full Goal acceptance or CI qualification. + After that bounded replan slice is acknowledged by a compact state run carrying `autonomous_replan_ack_v0` and `repair_delta_contract_v0` with `delta_present=true`, later empty monitor polls for the same acknowledged wait diff --git a/loopx/control_plane/goals/goal_frontier/__init__.py b/loopx/control_plane/goals/goal_frontier/__init__.py index d4d58520e8..be4a78c629 100644 --- a/loopx/control_plane/goals/goal_frontier/__init__.py +++ b/loopx/control_plane/goals/goal_frontier/__init__.py @@ -62,6 +62,7 @@ acceptance_gaps_from_held_goal_binding, ) from ...work_items.replan_semantics_codec import project_replan_semantics +from ...work_items.replan_history_codec import project_replan_history from .ack_policy import ( attach_rejected_replan_feedback, autonomous_replan_ack_satisfies_obligation, @@ -652,7 +653,7 @@ def _monitor_no_change_streak_trigger( if not isinstance(monitor_items, list): return None - stalled: list[tuple[int, str, dict[str, Any]]] = [] + eligible = [] for item in monitor_items: if not isinstance(item, dict): continue @@ -665,33 +666,17 @@ def _monitor_no_change_streak_trigger( claimed_by = agent_scope_item_claimed_by(item) if agent_id and claimed_by != agent_id: continue - no_change_count = safe_non_negative_int(item.get("consecutive_no_change")) - if no_change_count < MONITOR_NO_CHANGE_STREAK_THRESHOLD: - continue - target_key = str( - item.get("target_key") or item.get("todo_id") or "monitor" - ).strip() - stalled.append((no_change_count, target_key, item)) - if not stalled: + eligible.append(item) + if not eligible: return None - - no_change_count, target_key, monitor = min( - stalled, - key=lambda entry: (-entry[0], entry[1]), + # The enclosing frontier already established that no advancement is + # selectable. Keep its eligibility facts, but share count/expiry/winner + # selection and the existing public trigger with the typed replan owner. + return project_replan_history( + operation="frontier_monitor_streak", agent_id=agent_id, + agent_todos={"monitor_open_items": eligible}, + streak_threshold=MONITOR_NO_CHANGE_STREAK_THRESHOLD, ) - return { - "kind": MONITOR_NO_CHANGE_STREAK_TRIGGER, - "section": "agent_todo_summary.monitor_open_items", - "text": ( - f"monitor {target_key} recorded {no_change_count} consecutive " - "unchanged polls without selectable advancement" - ), - "todo_id": monitor.get("todo_id"), - "monitor_target_id": target_key, - "run_count": no_change_count, - "threshold": MONITOR_NO_CHANGE_STREAK_THRESHOLD, - "agent_id": agent_id, - } def _frontier_advancement_counts( diff --git a/loopx/control_plane/todos/summary_lanes.ts b/loopx/control_plane/todos/summary_lanes.ts index 0f7b030ce0..fa6ba9d3ee 100644 --- a/loopx/control_plane/todos/summary_lanes.ts +++ b/loopx/control_plane/todos/summary_lanes.ts @@ -37,7 +37,8 @@ export function decodeMonitorSchedule(row: JsonObject): MonitorSchedule { return {watchOnly: requireBoolean(row.watch_only, "watch_only"), dueAt: optionalTime(row.due_at, "due_at"), expiresAt: optionalTime(row.expires_at, "expires_at")}; } -const monitorIsActive = (row: MonitorSchedule, now: number) => row.expiresAt === null || row.expiresAt > now; +export const monitorIsActive = (row: Pick, now: number): boolean => + row.expiresAt === null || row.expiresAt > now; export function monitorIsDue(row: MonitorSchedule, now: number): boolean { return monitorIsActive(row, now) && row.dueAt !== null && row.dueAt <= now; } diff --git a/loopx/control_plane/work_items/replan_history.ts b/loopx/control_plane/work_items/replan_history.ts index c5e25af77c..f1df124936 100644 --- a/loopx/control_plane/work_items/replan_history.ts +++ b/loopx/control_plane/work_items/replan_history.ts @@ -6,6 +6,8 @@ import { requireStringArray, requireStringLiteral, optionalNonEmptyString, } from "../runtime_decode.ts"; import { projectTodoResumePlanning } from "../todos/resume_planning.ts"; +import { monitorIsActive } from "../todos/summary_lanes.ts"; +import { authorityUnicodeCompare } from "../coordination/authority_store_codec.ts"; export const PROGRESS_RESULT_CLASSES = [ "advanced", "unchanged", "blocked", "exploration_exhausted", "no_followup", @@ -51,7 +53,8 @@ interface Todos { resume: JsonObject | null; } interface Request { - operation: "all" | "progress" | "periodic" | "monitor_streak"; + operation: "all" | "progress" | "periodic" | "monitor_streak" | "frontier_monitor_streak"; + observedAt: number | null; agent: string | null; monitorAgent: string | null; runs: readonly Run[]; @@ -76,6 +79,8 @@ type Trigger = { run_count: number; threshold: number; agent_id: string | null } monitor_target_id: string; latest_generated_at: string } | { kind: "monitor_no_change_streak"; schema_version: string; section: "agent_todos"; monitor_target_id: string; text: string } + | { kind: "monitor_no_change_streak"; section: "agent_todo_summary.monitor_open_items"; + todo_id: string | null; monitor_target_id: string; text: string } ); function text(value: unknown, label: string): string { @@ -113,8 +118,16 @@ function decode(value: unknown): Request { const raw = requireJsonObject(value, "replan history"); requireStringLiteral(raw.schema_version, ["replan_history_request_v0"], "replan history schema"); const todo = requireJsonObject(raw.todos, "todos"); + const operation = requireStringLiteral(raw.operation, + ["all", "progress", "periodic", "monitor_streak", "frontier_monitor_streak"], "operation"); + // Historical requests have no current observation. Keep their replay + // behavior; the live frontier projection requires its runtime clock. + const observedAt = raw.observed_at === undefined ? null : time(raw.observed_at, "observed_at"); + if (operation === "frontier_monitor_streak" && observedAt === null) { + throw new EffectRuntimeRequestError("frontier monitor streak requires observed_at"); + } return { - operation: requireStringLiteral(raw.operation, ["all", "progress", "periodic", "monitor_streak"], "operation"), + operation, observedAt, agent: optionalNonEmptyString(raw.agent_id, "agent_id"), monitorAgent: optionalNonEmptyString(raw.monitor_agent_id, "monitor_agent_id"), neutral: new Set(requireStringArray(raw.neutral_classifications, "neutral_classifications")), @@ -293,10 +306,18 @@ function monitorTrigger(runs: readonly Run[], request: Request): Trigger | null }; } function monitorStreak(request: Request): Trigger | null { - const stalled = request.todos.monitors.filter(row => row.count >= request.streak) - .sort((a, b) => b.count - a.count)[0]; - if (!stalled || request.todos.advancements.some(row => row.status === "open" && + const frontier = request.operation === "frontier_monitor_streak"; + const stalled = request.todos.monitors.filter(row => row.count >= request.streak && + (request.observedAt === null || monitorIsActive({expiresAt: row.expiry}, request.observedAt))) + .sort((a, b) => b.count - a.count || (frontier ? authorityUnicodeCompare(a.target, b.target) : 0))[0]; + if (!stalled || !frontier && request.todos.advancements.some(row => row.status === "open" && row.taskClass === "advancement_task" && (!row.claim || row.claim === stalled.publicClaim))) return null; + if (frontier) return { + kind: "monitor_no_change_streak", section: "agent_todo_summary.monitor_open_items", + text: `monitor ${stalled.target} recorded ${stalled.count} consecutive unchanged polls without selectable advancement`, + todo_id: stalled.id, monitor_target_id: stalled.target, run_count: stalled.count, + threshold: request.streak, agent_id: request.agent, + }; return { kind: "monitor_no_change_streak", schema_version: request.monitorSchema, section: "agent_todos", text: `monitor ${stalled.target} recorded ${stalled.count} consecutive unchanged polls without runnable advancement`, @@ -313,7 +334,7 @@ export function projectReplanHistory(value: unknown, qualified?: ReadonlySet dict } -def _todo_facts(value: Any, *, include_resume: bool) -> dict[str, Any]: +def _todo_facts(value: Any, *, include_resume: bool, frontier: bool = False) -> dict[str, Any]: value = value if isinstance(value, dict) else {} raw_monitors = value.get("monitor_open_items") monitors = [] @@ -100,11 +100,14 @@ def _todo_facts(value: Any, *, include_resume: bool) -> dict[str, Any]: if not isinstance(item, dict): continue try: - count = int(str(item.get("consecutive_no_change") or "0")) - except ValueError: + raw_count = item.get("consecutive_no_change") or 0 + count = int(raw_count) if frontier else int(str(raw_count)) + except (TypeError, ValueError): count = 0 monitors.append({ - "id": normalize_todo_id(item.get("todo_id")), + # Frontier trigger identities are existing read labels, not a + # canonical execution lookup. Preserve the caller's public shape. + "id": item.get("todo_id") if frontier else normalize_todo_id(item.get("todo_id")), "claim": normalize_todo_claimed_by(item.get("claimed_by")), "public_claim": str(item.get("claimed_by") or "").strip() or None, "target": str(item.get("target_key") or item.get("todo_id") or "monitor").strip(), @@ -146,7 +149,7 @@ def _todo_facts(value: Any, *, include_resume: bool) -> dict[str, Any]: def project_replan_history( runs: Iterable[Mapping[str, Any]] = (), *, - operation: Literal["all", "progress", "periodic", "monitor_streak"] = "all", + operation: Literal["all", "progress", "periodic", "monitor_streak", "frontier_monitor_streak"] = "all", agent_id: str | None = None, agent_todos: Any = None, ack_recorded: Callable[..., bool] | None = None, neutral_classifications: set[str] | None = None, @@ -172,8 +175,13 @@ def project_replan_history( "stall_threshold": max(2, int(stall_threshold)), "periodic_threshold": periodic_threshold, "monitor_threshold": monitor_threshold, "streak_threshold": streak_threshold, "monitor_schema": monitor_schema, - "todos": _todo_facts(agent_todos, include_resume=needs_resume), + "todos": _todo_facts(agent_todos, include_resume=needs_resume, + frontier=operation == "frontier_monitor_streak"), } + if operation in {"monitor_streak", "frontier_monitor_streak"}: + # Carry one observation for the complete current source. Expiry is + # interpreted by the shared TypeScript Monitor owner, not this codec. + params["observed_at"] = now_utc().timestamp() if effective_turn_cadence is not None: params["settlement_source"] = effective_turn_cadence["settlement_source"] params["periodic_threshold"] = effective_turn_cadence["threshold"] diff --git a/tests/control_plane/test_monitor_replan_agent_scope.py b/tests/control_plane/test_monitor_replan_agent_scope.py index e7e45b06b5..a7202cc4d8 100644 --- a/tests/control_plane/test_monitor_replan_agent_scope.py +++ b/tests/control_plane/test_monitor_replan_agent_scope.py @@ -154,6 +154,36 @@ def test_peer_advancement_does_not_suppress_current_monitor_streak_replan() -> N } +def test_current_streak_keeps_eligibility_target_order_and_public_shape() -> None: + from copy import deepcopy + from loopx.control_plane.goals.goal_frontier import _monitor_no_change_streak_trigger + + def item(todo_id, target, **fields): + return quota_todo_item(todo_id=todo_id, title="Observe a finite source.", + task_class="continuous_monitor", claimed_by=AGENT_ID, + target_key=target, consecutive_no_change="8", expires_at="2099-01-01T00:00:00Z", + **fields) + + current = [item("todo_z", "z"), item("todo_a", "a")] + ignored = [ + {**current[0], "todo_id": "todo_peer", "claimed_by": PEER_AGENT_ID, "consecutive_no_change": "99"}, + {**current[0], "todo_id": "todo_unclaimed", "claimed_by": None, "consecutive_no_change": "99"}, + {**current[0], "watch_only": "true", "consecutive_no_change": "99"}, + {**current[0], "status": "blocked", "consecutive_no_change": "99"}, + {**current[0], "done": True, "consecutive_no_change": "99"}, + {**current[0], "task_class": "advancement_task", "consecutive_no_change": "99"}, + ] + source = {"monitor_open_items": ignored + current} + before = deepcopy(source) + assert _monitor_no_change_streak_trigger(source, agent_id=AGENT_ID) == { + "kind": "monitor_no_change_streak", "section": "agent_todo_summary.monitor_open_items", + "text": "monitor a recorded 8 consecutive unchanged polls without selectable advancement", + "todo_id": "todo_a", "monitor_target_id": "a", "run_count": 8, "threshold": 5, + "agent_id": AGENT_ID, + } + assert source == before + + def test_watch_only_monitor_streak_does_not_create_replan_obligation() -> None: monitor = quota_todo_item( todo_id="todo_watch_only_monitor", diff --git a/tests/control_plane/test_quota_monitor_clock.py b/tests/control_plane/test_quota_monitor_clock.py index fff0672c23..72605952ae 100644 --- a/tests/control_plane/test_quota_monitor_clock.py +++ b/tests/control_plane/test_quota_monitor_clock.py @@ -3,6 +3,8 @@ from copy import deepcopy from datetime import datetime, timedelta, timezone import json +import subprocess +import sys import pytest @@ -13,6 +15,7 @@ from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos from loopx.control_plane.todos import quota_selection from loopx.control_plane.todos.quota_summary import summarize_user_todos_for_quota +from loopx.control_plane.work_items import replan_history_codec from tests.control_plane.test_monitor_followthrough_contract import ( AGENT_ID, GOAL_ID, _add_monitor, _write_fixture, ) @@ -29,6 +32,7 @@ def monitor(todo_id: str, **fields: object) -> dict: def fixed_clock(monkeypatch) -> None: monkeypatch.setattr(monitor_todo, "now_utc", lambda: NOW) monkeypatch.setattr(quota_selection, "now_utc", lambda: NOW, raising=False) + monkeypatch.setattr(replan_history_codec, "now_utc", lambda: NOW, raising=False) @pytest.mark.parametrize("supported", [True, False]) @@ -130,6 +134,90 @@ def test_public_quota_gap_count_uses_real_full_source_without_writing_it(tmp_pat assert all(row.get("next_due_at") is None and row["status"] == "open" for row in records["todos"]) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("expiry, active", [ + ("2029-12-31T23:59:59Z", False), + ("2030-01-01T08:00:00+08:00", False), + ("2099-01-01T00:00:00Z", True), +]) +def test_codex_cli_replan_streak_respects_canonical_monitor_expiry( + tmp_path, monkeypatch, provider, expiry, active, +) -> None: + isolate_sqlite_runtime(tmp_path, monkeypatch) + runtime, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md" + state.write_text( + "---\nstatus: active\n---\n# Goal\n## Objective\nWait for an external result.\n\n## Agent Todo\n" + "- [ ] [P1] Observe a finite source.\n" + " \n" + "- [ ] [P1] Continue the result qualification when its input arrives.\n" + " \n", + encoding="utf-8", + ) + write_fixture_registry(project=tmp_path, runtime_root=runtime, registry_path=registry, + goal_id="goal-a", domain="monitor-clock", adapter_kind="generic_project_goal_v0", + state_file=str(state), registered_agents=["agent-a"], quota_allowed_slots=None) + goal = json.loads(registry.read_text())["goals"][0] + fields = parse_active_state_todos(state.read_text(), goal=goal, item_limit=None) + initialize_canonical_authority(runtime, "goal-a", build_todo_runtime_shadow_projection( + goal_id="goal-a", todos=fields["agent_todos"]["items"], handoff_mode="soft_claim"), + state_path=state, provider=provider) + before_state = state.read_bytes() + args = ("todo", "list", "--goal-id", "goal-a", "--role", "agent") + code, before = run_json_cli_result(*args, registry_path=registry, runtime_root=runtime) + assert code == 0, before + # Freeze the observation inside the real CLI process, rather than assuming + # that parent-process monkeypatches cross a subprocess boundary. Stores, + # command parsing and the TypeScript runtime remain the production path. + entrypoint = "\n".join([ + "from datetime import datetime, timezone", + "from loopx.control_plane.scheduler import monitor_todo", + "from loopx.control_plane.todos import quota_selection", + "from loopx.control_plane.work_items import replan_history_codec", + "clock = lambda: datetime(2030, 1, 1, tzinfo=timezone.utc)", + "monitor_todo.now_utc = clock", + "quota_selection.now_utc = clock", + "replan_history_codec.now_utc = clock", + "from loopx.cli import main", + "raise SystemExit(main())", + ]) + result = subprocess.run([ + sys.executable, "-c", entrypoint, "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json", + "quota", "should-run", "--codex-app", "--goal-id", "goal-a", "--agent-id", "agent-a", + "--turn-instance-id", "expiry-fixture-turn", "--scan-path", str(tmp_path), + ], capture_output=True, text=True, encoding="utf-8", check=False) + assert result.returncode == 0, result.stdout + result.stderr + guard = json.loads(result.stdout) + kinds = [row["kind"] for row in (guard.get("autonomous_replan_obligation") or {}).get("triggers", [])] + assert ("monitor_no_change_streak" in kinds) is active, guard + if not active: + assert guard["should_run"] is False, guard + assert guard["agent_todo_summary"]["monitor_due_count"] == 0 + code, after = run_json_cli_result(*args, registry_path=registry, runtime_root=runtime) + assert code == 0, after + # Resume evaluation time is a read observation; persisted identities and + # counters must remain unchanged across the actual Codex CLI admission. + def stable(rows): + return [{k: v for k, v in row.items() if k != "resume_condition"} for row in rows] + assert stable(after["todos"]) == stable(before["todos"]) + assert state.read_bytes() == before_state + index = runtime / "goals" / "goal-a" / "runs" / "index.jsonl" + events = [json.loads(line) for line in index.read_text().splitlines()] if index.exists() else [] + assert not any(row.get("classification") == "quota_slot_spent" for row in events) + # A quiet guard may append its liveness receipt; it cannot execute a + # source observation or advance that source's persisted counters. + for row in events: + if row.get("classification") == "quota_monitor_poll": + assert row.get("todo_id") is None + assert row.get("monitor_event") is None + assert row["monitor_target"]["monitor_mode"] == "monitor_quiet_until_material_transition" + + def test_missing_typed_quota_owner_cannot_reactivate_python_monitor_rules(monkeypatch) -> None: def unavailable(*_args, **_kwargs): raise RuntimeError("isolated typed owner unavailable") diff --git a/tests/control_plane/test_replan_history_policy.py b/tests/control_plane/test_replan_history_policy.py index 777b7d0249..9b57536cfa 100644 --- a/tests/control_plane/test_replan_history_policy.py +++ b/tests/control_plane/test_replan_history_policy.py @@ -185,6 +185,35 @@ def test_long_history_keeps_evidence_beyond_rpc_limit() -> None: assert trigger["oldest_counted_generated_at"] == rows[-1]["generated_at"] +def test_streak_codec_carries_one_clock_for_all_monitor_facts(monkeypatch) -> None: + from datetime import datetime, timezone + from loopx.control_plane.work_items import replan_history_codec as codec + + calls, clocks = [], [] + observed = datetime(2030, 1, 1, tzinfo=timezone.utc) + + def clock(): + clocks.append(True) + return observed + + def capture(method, params): + calls.append((method, params)) + return {"schema_version": "replan_history_result_v0", "trigger": None} + + monkeypatch.setattr(codec, "now_utc", clock) + monkeypatch.setattr(codec, "effect_runtime_result", capture) + items = [{"todo_id": f"todo_monitor_{i}", "claimed_by": AGENT, + "consecutive_no_change": "8", "expires_at": "2030-01-01T00:00:00Z"} + for i in range(8)] + codec.project_replan_history(operation="monitor_streak", agent_todos={"monitor_open_items": items}) + assert clocks == [True] + assert len(calls) == 1 and calls[0][0] == "work_item.replan_history.project" + params = calls[0][1] + assert params["observed_at"] == observed.timestamp() + assert len(params["todos"]["monitors"]) == 8 + assert all(row["expires_at"] == params["observed_at"] for row in params["todos"]["monitors"]) + + def test_snapshot_transport_is_private_compact_and_always_removed(monkeypatch) -> None: import json import os diff --git a/tests/control_plane_ts/replan_history.test.ts b/tests/control_plane_ts/replan_history.test.ts index a680e7d001..7274cacfc9 100644 --- a/tests/control_plane_ts/replan_history.test.ts +++ b/tests/control_plane_ts/replan_history.test.ts @@ -180,6 +180,52 @@ test("persisted monitor streak retains stable tie order and advancement ownershi } }), null); }); +test("current monitor streak ignores expiry at or before its observation, without changing live streaks", () => { + const source = request([], {operation: "monitor_streak", observed_at: 100, + todos: {monitors: [], advancements: [], resume: null}}); + for (const [expiry, due, count, expected] of [ + [99, 90, 8, false], [100, 90, 8, false], [101, 90, 8, true], + [null, 90, 8, true], [200, 110, 8, true], [200, 90, 4, false], + ] as const) { + const value = {...source, todos: {monitors: [monitorTodo({expires_at: expiry, + due_at: due, no_change_count: count})], advancements: [], resume: null}}; + const before = structuredClone(value); + assert.equal(projectReplanHistory(value).trigger !== null, expected, + `expiry=${expiry}, due=${due}, count=${count}`); + assert.deepEqual(value, before); + } + const result = projectReplanHistory({...source, todos: {advancements: [], resume: null, + monitors: [monitorTodo({target: "retired", expires_at: 100, no_change_count: 99}), + monitorTodo({target: "active", expires_at: 101, no_change_count: 5})]}}); + assert.equal((result.trigger as JsonObject).monitor_target_id, "active"); +}); + +test("legacy replan requests retain their clockless behavior and independent history triggers", () => { + const todos = {monitors: [monitorTodo({expires_at: 0})], advancements: [], resume: null}; + assert.equal(trigger([], {operation: "monitor_streak", todos})?.kind, "monitor_no_change_streak"); + assert.equal(trigger(many(20), {observed_at: 100, todos})?.kind, "periodic_review_due"); + assert.equal(trigger([run(2, {progress: observation}), run(1, {progress: observation})], + {observed_at: 100, todos})?.kind, "typed_progress_repeat"); + for (const clock of [NaN, Infinity, "100"]) { + assert.throws(() => projectReplanHistory(request([], {observed_at: clock})), /observed_at/); + } +}); + +test("the current frontier uses the same expiry owner and preserves its public trigger and target tie order", () => { + const todos = {monitors: [monitorTodo({target: "retired", expires_at: 100, no_change_count: 99}), + monitorTodo({id: "todo_z", target: "z", expires_at: 101, no_change_count: 8}), + monitorTodo({id: "todo_a", target: "a", expires_at: 101, no_change_count: 8})], + advancements: [], resume: null}; + assert.deepEqual(trigger([], {operation: "frontier_monitor_streak", observed_at: 100, todos}), { + kind: "monitor_no_change_streak", section: "agent_todo_summary.monitor_open_items", + text: "monitor a recorded 8 consecutive unchanged polls without selectable advancement", + todo_id: "todo_a", monitor_target_id: "a", run_count: 8, threshold: 5, agent_id: "worker-a", + }); + for (const observed_at of [undefined, null]) { + assert.throws(() => trigger([], {operation: "frontier_monitor_streak", observed_at, todos}), /observed_at/); + } +}); + for (const [key, bad, message] of [ ["schema_version", "future_schema", /schema/], ["operation", "dispatch", /operation/], ["runs", {}, /runs/], ["stall_threshold", 1, /stall_threshold/],