From f38d5ca8565b74849d889d3ad755f405fd89bc77 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 9 Sep 2026 23:56:03 +0800 Subject: [PATCH 1/3] fix(monitor): reduce observation state in the typed Todo field plan Signed-off-by: huangruiteng --- .../issue_fix/pr_monitor_materialization.py | 18 +- .../control_plane/effect_runtime_handlers.ts | 2 + loopx/control_plane/runtime_timestamp.ts | 86 +++++ loopx/control_plane/scheduler/monitor_todo.py | 7 - loopx/control_plane/todos/field_update.ts | 20 +- loopx/control_plane/todos/line_update.py | 11 +- loopx/control_plane/todos/monitor_metadata.py | 339 ++---------------- loopx/control_plane/todos/monitor_metadata.ts | 175 +++++++++ .../work_items/task_lease_acquire.ts | 41 +-- loopx/todos.py | 32 +- 10 files changed, 334 insertions(+), 397 deletions(-) create mode 100644 loopx/control_plane/runtime_timestamp.ts create mode 100644 loopx/control_plane/todos/monitor_metadata.ts diff --git a/loopx/capabilities/issue_fix/pr_monitor_materialization.py b/loopx/capabilities/issue_fix/pr_monitor_materialization.py index 32cc81d479..f5b5fd6bcf 100644 --- a/loopx/capabilities/issue_fix/pr_monitor_materialization.py +++ b/loopx/capabilities/issue_fix/pr_monitor_materialization.py @@ -8,8 +8,8 @@ from ...control_plane.scheduler.monitor_todo import ( monitor_next_due_at, - parse_monitor_counter, ) +from ...control_plane.todos.monitor_metadata import MonitorPollObservation from ...todos import ( add_goal_todo, complete_goal_todo, @@ -145,18 +145,13 @@ def materialize_issue_fix_grouped_monitors( previous_hash = str((previous or {}).get("result_hash") or "") reopening = bool((previous or {}).get("done")) material_change = reopening or previous_hash != result_hash - previous_no_change = parse_monitor_counter( - (previous or {}).get("consecutive_no_change") - ) monitor_metadata = { "target_key": target_key, "cadence": cadence, "next_due_at": next_due_at, "last_checked_at": generated_at, "result_hash": result_hash, - "consecutive_no_change": ( - "0" if material_change else str(previous_no_change + 1) - ), + "consecutive_no_change": "0", "material_change": "true" if material_change else "false", "watch_only": "true", } @@ -185,7 +180,14 @@ def materialize_issue_fix_grouped_monitors( role="agent", status="open" if reopening else None, reason=reason, - monitor_metadata=monitor_metadata, + # Membership is an observation, not precomputed Todo state. + # The writer derives counters/generation against its locked + # snapshot, just like quota monitor-poll. + monitor_metadata=MonitorPollObservation( + generated_at=generated_at, result_hash=result_hash, + material_change=material_change, target_key=target_key, + cadence=cadence, next_due_at=next_due_at, + ), no_followup=False if reopening else None, agent_id=claimed_by, project=project, diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 0393a210bb..ac03f744cc 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -66,6 +66,7 @@ import { import { reduceTodoCompletionTransaction } from "./todos/completion_transaction.ts"; import { transitionTodoNextAction } from "./todos/next_action.ts"; import { planTodoFieldUpdate } from "./todos/field_update.ts"; +import { planMonitorMetadata } from "./todos/monitor_metadata.ts"; import { planTodoAuthoringScope } from "./todos/authoring_scope.ts"; import { evaluateTodoResumeConditions, @@ -370,6 +371,7 @@ export function createEffectRuntimeHandlers( ["todo.completion_state.require_metadata", requireTodoCompletionMetadataValue], ["todo.completion_state.continuation_for_write", selectTodoCompletionContinuation], ["todo.field_update.plan", planTodoFieldUpdate], + ["todo.monitor_metadata.plan", planMonitorMetadata], ["todo.authoring_scope.plan", planTodoAuthoringScope], [ "todo.claim.decide", diff --git a/loopx/control_plane/runtime_timestamp.ts b/loopx/control_plane/runtime_timestamp.ts new file mode 100644 index 0000000000..80b7e3668d --- /dev/null +++ b/loopx/control_plane/runtime_timestamp.ts @@ -0,0 +1,86 @@ +/** Calendar-checked ISO codec; Date.parse alone silently rolls invalid dates. */ +export function parseIsoTimestamp(value: string): Date | null { + const match = /^(\d{4})-(\d{2})-(\d{2})(?:[T ](\d{2}):(\d{2})(?::(\d{2})(?:\.(\d+))?)?(Z|z|[+-]\d{2}(?::?\d{2})?)?)?$/u.exec( + value.trim(), + ); + if (match === null) return null; + const [, yearText, monthText, dayText, hourText, minuteText, secondText, fraction, timezone] = match; + const [year, month, day, hour, minute, second, millisecond] = [ + yearText, + monthText, + dayText, + hourText ?? "0", + minuteText ?? "0", + secondText ?? "0", + (fraction ?? "").slice(0, 3).padEnd(3, "0") || "0", + ].map(Number); + const endOfDay = hour === 24; + if ( + endOfDay && + (minute !== 0 || second !== 0 || (fraction !== undefined && /[1-9]/u.test(fraction))) + ) return null; + const calendarHour = endOfDay ? 0 : hour; + const calendar = new Date(0); + calendar.setUTCHours(calendarHour, minute, second, millisecond); + calendar.setUTCFullYear(year, month - 1, day); + if ( + calendar.getUTCFullYear() !== year || calendar.getUTCMonth() !== month - 1 || + calendar.getUTCDate() !== day || calendar.getUTCHours() !== calendarHour || + calendar.getUTCMinutes() !== minute || calendar.getUTCSeconds() !== second || + calendar.getUTCMilliseconds() !== millisecond + ) return null; + if (hourText === undefined) return calendar; + let text = value.trim().replace(" ", "T").replace(/z$/u, "Z"); + if (fraction !== undefined) text = text.replace(`.${fraction}`, `.${fraction.slice(0, 3)}`); + if (timezone === undefined) text += "Z"; + else text = text.replace(/([+-]\d{2})$/u, "$1:00"); + const parsed = new Date(text); + return Number.isNaN(parsed.valueOf()) ? null : parsed; +} + +/** Compatibility codec for datetime.fromisoformat inputs used by Todo metadata. + * It keeps microseconds and offset seconds, which a JS Date cannot represent. + * Missing timezone means UTC, matching the existing Python runtime codec. */ +export function parseTodoTimestampMicros(value: string): bigint | null { + const match = /^(\d{4}-\d{2}-\d{2}|\d{8}|\d{4}-W\d{2}(?:-[1-7])?|\d{4}W\d{2}[1-7]?)(?:[\s\S](.+))?$/u.exec(value); + if (!match) return null; + const [, date, time] = match; + let calendar: Date | null; + if (date.includes("W")) { + const week = /^(\d{4})-?W(\d{2})(?:-?([1-7]))?$/.exec(date)!; + const year = Number(week[1]), number = Number(week[2]), day = Number(week[3] ?? 1); + if (year < 1 || number < 1 || number > 53) return null; + calendar = parseIsoTimestamp(`${week[1]}-01-04`); + if (!calendar) return null; + calendar.setUTCDate(calendar.getUTCDate() - (calendar.getUTCDay() + 6) % 7 + (number - 1) * 7 + day - 1); + const thursday = new Date(calendar); + thursday.setUTCDate(thursday.getUTCDate() + 3 - (thursday.getUTCDay() + 6) % 7); + if (thursday.getUTCFullYear() !== year || calendar.getUTCFullYear() > 9999) return null; + } else { + const expanded = date.includes("-") ? date : `${date.slice(0, 4)}-${date.slice(4, 6)}-${date.slice(6)}`; + calendar = parseIsoTimestamp(expanded); + if (!calendar || calendar.getUTCFullYear() < 1) return null; + } + if (time === undefined) return BigInt(calendar.valueOf()) * 1000n; + const parts = /^(.*?)(Z|z|[+-].*)?$/.exec(time)!; + function clock(raw: string, offset: boolean): bigint | null { + const parsed = /^(\d{2})(?:(:?)?(\d{2})(?:\2(\d{2}))?)?(?:[.,](\d+))?$/.exec(raw); + if (!parsed) return null; + const hour = Number(parsed[1]), minute = Number(parsed[3] ?? 0), second = Number(parsed[4] ?? 0); + if (!offset && (hour > 23 || minute > 59 || second > 59)) return null; + const seconds = hour * 3600 + minute * 60 + second; + const micros = BigInt(seconds) * 1000000n + BigInt((parsed[5] ?? "").padEnd(6, "0").slice(0, 6)); + if (offset && micros >= 86400000000n) return null; + // Python treats an all-zero offset as UTC even with fractional seconds. + return offset && seconds === 0 ? 0n : micros; + } + const local = clock(parts[1], false); + if (local === null) return null; + let offset = 0n; + if (parts[2] && !["Z", "z"].includes(parts[2])) { + const parsed = clock(parts[2].slice(1), true); + if (parsed === null) return null; + offset = parts[2][0] === "-" ? -parsed : parsed; + } + return BigInt(calendar.valueOf()) * 1000n + local - offset; +} diff --git a/loopx/control_plane/scheduler/monitor_todo.py b/loopx/control_plane/scheduler/monitor_todo.py index e78237b864..eb8870b4c4 100644 --- a/loopx/control_plane/scheduler/monitor_todo.py +++ b/loopx/control_plane/scheduler/monitor_todo.py @@ -19,13 +19,6 @@ parse_monitor_timestamp = parse_scheduler_timestamp -def parse_monitor_counter(value: Any) -> int: - try: - return max(0, int(str(value or "0").strip())) - except ValueError: - return 0 - - def monitor_cadence_delta(value: Any) -> timedelta | None: projected = project_monitor_todo_schedule( generated_at="1970-01-01T00:00:00Z", diff --git a/loopx/control_plane/todos/field_update.ts b/loopx/control_plane/todos/field_update.ts index 8ba4e2098e..cff9664063 100644 --- a/loopx/control_plane/todos/field_update.ts +++ b/loopx/control_plane/todos/field_update.ts @@ -10,6 +10,7 @@ import { TODO_COMPLETION_STATE_REQUEST_SCHEMA, } from "./completion_state.ts"; import { normalizeTodoResumeWhen, TODO_RESUME_NORMALIZE_REQUEST_SCHEMA_VERSION } from "./resume_condition.ts"; +import { MONITOR_METADATA_FIELDS, planMonitorMetadata, TODO_MONITOR_METADATA_REQUEST_SCHEMA } from "./monitor_metadata.ts"; export const TODO_FIELD_UPDATE_REQUEST_SCHEMA = "loopx_todo_field_update_request_v0"; export const TODO_FIELD_UPDATE_RESULT_SCHEMA = "loopx_todo_field_update_result_v0"; @@ -27,9 +28,6 @@ const STRING_FIELDS = ["note", "evidence", "completion_turn_key", "reason", "tas const PRESENT_FIELDS = ["required_write_scopes", "required_capabilities", "target_capabilities", "explore_result_node_refs", "decision_scope", "required_decision_scopes", "decision_outcome", "decision_scope_outcomes"] as const; -const MONITOR_FIELDS = ["target_key", "monitor_effect_id", "cadence", "next_due_at", "expires_at", - "last_checked_at", "result_hash", "consecutive_no_change", "material_change", - "material_change_generation", "max_no_change_before_replan", "watch_only"] as const; const FLAGS = ["clear_claim", "claim_only", "clear_user_binding", "clear_blocks_agent", "clear_global_gate", "clear_resume_when"] as const; const INTENT_FIELDS = new Set([...STRING_FIELDS, ...PRESENT_FIELDS, ...FLAGS, @@ -190,10 +188,18 @@ export function planTodoFieldUpdate(value: unknown): TodoFieldUpdatePlan { } if (present(intent.no_followup)) updates.no_followup = intent.no_followup; Object.assign(updates, completionUpdates(block, intent, targetStatus, normalizedStatus)); - if (present(intent.monitor_metadata)) { - const monitor = requireJsonObject(intent.monitor_metadata, "monitor metadata"); - for (const field of MONITOR_FIELDS) if (Object.hasOwn(monitor, field)) updates[field] = monitor[field]; + // Public update carries the effective scope and raw observation once. The + // field plan composes validation and generation without another RPC. + const monitorPlan = request.monitor_context == null ? null : planMonitorMetadata({ + ...requireJsonObject(request.monitor_context, "monitor context"), + schema_version: TODO_MONITOR_METADATA_REQUEST_SCHEMA, existing: block, generated_at: updatedAt, + }); + const monitor = monitorPlan?.metadata ?? intent.monitor_metadata; + if (present(monitor)) { + const metadata = requireJsonObject(monitor, "monitor metadata"); + for (const field of MONITOR_METADATA_FIELDS) if (Object.hasOwn(metadata, field)) updates[field] = metadata[field]; } return {schema_version: TODO_FIELD_UPDATE_RESULT_SCHEMA, normalized_status: normalizedStatus, - target_status: targetStatus, metadata_updates: updates}; + target_status: targetStatus, metadata_updates: updates, + ...(monitorPlan?.transition ? {monitor_poll_transition: monitorPlan.transition} : {})}; } diff --git a/loopx/control_plane/todos/line_update.py b/loopx/control_plane/todos/line_update.py index 54a399c777..876a56ed1c 100644 --- a/loopx/control_plane/todos/line_update.py +++ b/loopx/control_plane/todos/line_update.py @@ -42,6 +42,7 @@ normalize_todo_completion_continuation, normalize_todo_completion_recovery, ) +from .contract import TODO_MONITOR_METADATA_FIELDS def upsert_todo_metadata( @@ -150,7 +151,8 @@ def link_superseding_todo_id( def _field_update_plan( - block: Mapping[str, Any], intent: dict[str, Any], updated_at: str + block: Mapping[str, Any], intent: dict[str, Any], updated_at: str, + monitor_context: dict[str, Any] | None = None, ) -> dict[str, Any]: """Adapt source facts only; the TS planner owns omission/clear/state rules.""" try: @@ -169,10 +171,13 @@ def _field_update_plan( "no_followup", "completion_continuation", "successor_todo_ids", + "task_class", + *TODO_MONITOR_METADATA_FIELDS, ) }, "intent": intent, "updated_at": updated_at, + "monitor_context": monitor_context, }, ) except EffectRuntimeRejected as exc: @@ -232,6 +237,7 @@ def apply_todo_update_to_lines( clear_resume_when: bool = False, no_followup: bool | None = None, monitor_metadata: dict[str, Any] | None = None, + monitor_context: dict[str, Any] | None = None, clear_claim: bool = False, claim_only: bool = False, updated_at: str, @@ -299,6 +305,7 @@ def apply_todo_update_to_lines( "claim_only": claim_only, }, updated_at, + monitor_context, ) normalized_status = plan["normalized_status"] target_status = plan["target_status"] @@ -319,6 +326,8 @@ def apply_todo_update_to_lines( metadata_updated = upsert_todo_metadata(lines, block, metadata_line) effective_metadata = parse_todo_metadata_line(metadata_line or "") or {} return { + **({"monitor_poll_transition": plan["monitor_poll_transition"]} + if "monitor_poll_transition" in plan else {}), "role": resolved_role, "section": section, "todo": block.get("text"), diff --git a/loopx/control_plane/todos/monitor_metadata.py b/loopx/control_plane/todos/monitor_metadata.py index 65797207e1..78fb6f8be7 100644 --- a/loopx/control_plane/todos/monitor_metadata.py +++ b/loopx/control_plane/todos/monitor_metadata.py @@ -1,25 +1,15 @@ +"""Monitor wire adaptation. State rules live in the TS Todo field planner.""" from __future__ import annotations -from collections.abc import Mapping -from dataclasses import dataclass +from dataclasses import asdict, dataclass from typing import Any -from ..runtime.time import parse_timestamp -from ..scheduler.monitor_todo import ( - monitor_cadence_delta, - monitor_next_due_at, - parse_monitor_counter, -) -from .contract import ( - TODO_MONITOR_METADATA_FIELDS, - TODO_TASK_CLASS_MONITOR, - normalize_todo_watch_only, -) +from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result @dataclass(frozen=True) class MonitorPollObservation: - """One monitor result whose state transition must be planned under lock.""" + """One observation, reduced against the Todo state under its writer lock.""" generated_at: str result_hash: str @@ -33,306 +23,31 @@ class MonitorPollObservation: MonitorMetadataInput = dict[str, Any] | MonitorPollObservation | None -def plan_monitor_poll_metadata( - *, - existing: Mapping[str, Any], - observation: MonitorPollObservation, -) -> tuple[dict[str, Any], dict[str, Any]]: - """Derive one monotonic monitor transition from the locked Todo state.""" - - if str(existing.get("task_class") or "") != TODO_TASK_CLASS_MONITOR: - raise ValueError( - "monitor poll observation requires task_class=continuous_monitor" - ) - result_hash = str(observation.result_hash or "").strip() - if not result_hash: - raise ValueError("monitor todo writeback requires --result-hash") - - existing_target_key = str(existing.get("target_key") or "").strip() - requested_target_key = str(observation.target_key or "").strip() - if ( - requested_target_key - and existing_target_key - and requested_target_key != existing_target_key - ): - raise ValueError( - f"monitor poll target_key resolves to {existing_target_key!r}, " - f"not {requested_target_key!r}" - ) - target_key = requested_target_key or existing_target_key - cadence = str(observation.cadence or existing.get("cadence") or "").strip() - next_due_at = monitor_next_due_at( - generated_at=observation.generated_at, - cadence=cadence, - explicit_next_due_at=observation.next_due_at, - ) - if not observation.material_change and not next_due_at: - raise ValueError( - "unchanged monitor todo writeback requires --next-due-at or a " - "parseable cadence such as 30m/2h/1d" - ) - - monitor_effect_id = str(observation.monitor_effect_id or "").strip() - existing_effect_id = str(existing.get("monitor_effect_id") or "").strip() - if monitor_effect_id and monitor_effect_id == existing_effect_id: - replay_facts = { - "result_hash": result_hash, - "material_change": "true" if observation.material_change else "false", - "last_checked_at": observation.generated_at, - "target_key": target_key, - "cadence": cadence, - "next_due_at": next_due_at or "", - } - conflicts = [ - key - for key, expected in replay_facts.items() - if str(existing.get(key) or "").strip() != str(expected or "").strip() - ] - if conflicts: - raise ValueError( - "monitor effect identity is already bound to different " - f"observation fields: {', '.join(conflicts)}" - ) - existing_metadata = { - key: existing[key] - for key in TODO_MONITOR_METADATA_FIELDS - if existing.get(key) is not None - } - return existing_metadata, { - "monitor_effect_id": monitor_effect_id, - "provider_replayed": True, - "result_hash": result_hash, - "material_change": observation.material_change, - "material_change_applied": False, - "material_change_generation": parse_monitor_counter( - existing.get("material_change_generation") - ), - "consecutive_no_change": parse_monitor_counter( - existing.get("consecutive_no_change") - ), - "last_checked_at": observation.generated_at, - "target_key": target_key or None, - "cadence": cadence or None, - "next_due_at": next_due_at, - } - - if monitor_effect_id and existing_effect_id: - persisted_at = parse_timestamp(existing.get("last_checked_at")) - observed_at = parse_timestamp(observation.generated_at) - if persisted_at is not None and observed_at is not None and observed_at <= persisted_at: - raise ValueError( - "monitor observation is older than the persisted monitor effect" - ) - - previous_hash = str(existing.get("result_hash") or "").strip() - previous_no_change = parse_monitor_counter(existing.get("consecutive_no_change")) - previous_generation = parse_monitor_counter( - existing.get("material_change_generation") - ) - advances_generation = bool( - observation.material_change and result_hash != previous_hash +def monitor_metadata_intent(value: MonitorMetadataInput) -> dict[str, Any]: + return ( + {"observation": asdict(value), "metadata": None} + if isinstance(value, MonitorPollObservation) + else {"observation": None, "metadata": value} ) - generation = previous_generation + (1 if advances_generation else 0) - consecutive_no_change = ( - 0 - if observation.material_change - or (previous_hash and previous_hash != result_hash) - else previous_no_change + 1 - ) - - metadata: dict[str, Any] = { - "last_checked_at": observation.generated_at, - "result_hash": result_hash, - "consecutive_no_change": str(consecutive_no_change), - "material_change": "true" if observation.material_change else "false", - "material_change_generation": str(generation), - } - if monitor_effect_id: - metadata["monitor_effect_id"] = monitor_effect_id - if target_key: - metadata["target_key"] = target_key - if cadence: - metadata["cadence"] = cadence - if next_due_at: - metadata["next_due_at"] = next_due_at - return metadata, { - "monitor_effect_id": monitor_effect_id or None, - "provider_replayed": False, - "result_hash": result_hash, - "material_change": observation.material_change, - "material_change_applied": advances_generation, - "material_change_generation": generation, - "consecutive_no_change": consecutive_no_change, - "last_checked_at": observation.generated_at, - "target_key": target_key or None, - "cadence": cadence or None, - "next_due_at": next_due_at, - } - - -def resolve_monitor_metadata_input( - *, - existing: Mapping[str, Any], - monitor_metadata: MonitorMetadataInput, -) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: - if not isinstance(monitor_metadata, MonitorPollObservation): - return monitor_metadata, None - return plan_monitor_poll_metadata( - existing=existing, - observation=monitor_metadata, - ) - - -def normalize_monitor_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]: - normalized: dict[str, Any] = {} - for key, value in (metadata or {}).items(): - if key not in TODO_MONITOR_METADATA_FIELDS: - continue - if value is None: - normalized[key] = None - continue - candidate = str(value or "").strip() - if candidate: - normalized[key] = candidate - if ( - normalized.get("cadence") is not None - and monitor_cadence_delta(normalized["cadence"]) is None - ): - raise ValueError("--cadence must look like 30m, 2h, or 1d") - if normalized.get("next_due_at") is not None and parse_timestamp(normalized["next_due_at"]) is None: - raise ValueError("--next-due-at must be an ISO timestamp") - if normalized.get("expires_at") is not None and parse_timestamp(normalized["expires_at"]) is None: - raise ValueError("--expires-at must be an ISO timestamp") - if normalized.get("last_checked_at") is not None and parse_timestamp(normalized["last_checked_at"]) is None: - raise ValueError("--last-checked-at must be an ISO timestamp") - if normalized.get("consecutive_no_change") is not None: - try: - int(normalized["consecutive_no_change"]) - except ValueError as exc: - raise ValueError("--consecutive-no-change must be an integer") from exc - if normalized.get("material_change_generation") is not None: - try: - generation = int(normalized["material_change_generation"]) - except ValueError as exc: - raise ValueError( - "--material-change-generation must be an integer" - ) from exc - if generation < 0: - raise ValueError( - "--material-change-generation must be a non-negative integer" - ) - if normalized.get("material_change") is not None and normalized["material_change"] not in {"true", "false"}: - raise ValueError("--material-change metadata must be true or false") - if normalized.get("watch_only") is not None and normalize_todo_watch_only( - normalized["watch_only"] - ) is None: - raise ValueError("--watch-only metadata must be true or false") - return normalized - - -def materialize_monitor_schedule( - *, - task_class: str | None, - monitor_metadata: dict[str, Any], - generated_at: str, -) -> dict[str, Any]: - """Materialize the first due time for a cadence-only monitor mutation.""" - - if task_class != TODO_TASK_CLASS_MONITOR: - return monitor_metadata - if monitor_metadata.get("next_due_at") is not None: - return monitor_metadata - cadence = monitor_metadata.get("cadence") - if cadence is None: - return monitor_metadata - next_due_at = monitor_next_due_at( - generated_at=generated_at, - cadence=cadence, - ) - if next_due_at is None: - return monitor_metadata - return {**monitor_metadata, "next_due_at": next_due_at} - - -def require_continuous_monitor_boundedness( - *, - task_class: str | None, - resume_when: str | None, - monitor_metadata: dict[str, Any] | None, -) -> None: - if task_class != TODO_TASK_CLASS_MONITOR: - return - metadata = monitor_metadata or {} - if ( - str(metadata.get("expires_at") or "").strip() - or str(resume_when or "").strip() - or normalize_todo_watch_only(metadata.get("watch_only")) is True - ): - return - raise ValueError( - "continuous_monitor requires one of: --expires-at, --resume-when, " - "or --watch-only" - ) - - -def validate_monitor_metadata_update( - *, - monitor_metadata: dict[str, Any] | None, - existing: Mapping[str, Any], - role: str, - task_class: str | None, - generated_at: str, - resume_when: str | None, - enforce_boundedness: bool, -) -> dict[str, Any]: - normalized = require_monitor_metadata_scope( - monitor_metadata=monitor_metadata, - role=role, - task_class=task_class, - generated_at=generated_at, - ) - if enforce_boundedness: - effective = { - key: value - for key, value in { - **{ - key: existing.get(key) - for key in ("expires_at", "watch_only") - if existing.get(key) is not None - }, - **normalized, - }.items() - if value is not None - } - require_continuous_monitor_boundedness( - task_class=task_class, - resume_when=resume_when, - monitor_metadata=effective, - ) - return normalized def require_monitor_metadata_scope( - *, - monitor_metadata: dict[str, Any] | None, - role: str, - task_class: str | None, - generated_at: str | None = None, + *, monitor_metadata: dict[str, Any] | None, role: str, + task_class: str | None, generated_at: str | None = None, + resume_when: str | None = None, enforce_boundedness: bool = False, ) -> dict[str, Any]: - normalized = normalize_monitor_metadata(monitor_metadata) - if not normalized: - return {} - schedule_fields = {k for k, v in normalized.items() if v is not None and k != "target_key"} - if schedule_fields and (role != "agent" or task_class != "continuous_monitor"): - raise ValueError( - "monitor schedule metadata requires --role agent --task-class continuous_monitor" - ) - if normalized.get("target_key") is not None and role != "agent": - raise ValueError("target_key requires --role agent") - if generated_at is None: - return normalized - return materialize_monitor_schedule( - task_class=task_class, - monitor_metadata=normalized, - generated_at=generated_at, - ) + """Create/line-codec adapter; update composes this owner in its field plan.""" + try: + result = effect_runtime_result("todo.monitor_metadata.plan", { + "schema_version": "loopx_todo_monitor_metadata_request_v0", + "existing": {}, "metadata": monitor_metadata, "role": role, + "task_class": task_class, "generated_at": generated_at, + "resume_when": resume_when, "enforce_boundedness": enforce_boundedness, + }) + except EffectRuntimeRejected as exc: + raise ValueError(str(exc)) from None + if (not isinstance(result, dict) + or result.get("schema_version") != "loopx_todo_monitor_metadata_result_v0" + or not isinstance(result.get("metadata"), dict)): + raise RuntimeError("TypeScript monitor metadata result shape mismatch") + return result["metadata"] diff --git a/loopx/control_plane/todos/monitor_metadata.ts b/loopx/control_plane/todos/monitor_metadata.ts new file mode 100644 index 0000000000..e08d689c34 --- /dev/null +++ b/loopx/control_plane/todos/monitor_metadata.ts @@ -0,0 +1,175 @@ +/** Monitor authoring and observation rules. A plan is not a commit receipt: + * the caller must hold the Todo writer lock and retain its authority fence. */ +import type { JsonObject } from "../effect_program.ts"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { requireBoolean, requireJsonObject } from "../runtime_decode.ts"; +import { stripPythonWhitespace } from "../coordination/todo_agents.ts"; +import { evaluateSchedulerStateTransition, SCHEDULER_STATE_TRANSITION_REQUEST_SCHEMA } from "../scheduler/state_transition_rules.ts"; +import { parseTodoTimestampMicros } from "../runtime_timestamp.ts"; + +export const TODO_MONITOR_METADATA_REQUEST_SCHEMA = "loopx_todo_monitor_metadata_request_v0"; +export const TODO_MONITOR_METADATA_RESULT_SCHEMA = "loopx_todo_monitor_metadata_result_v0"; +export const MONITOR_METADATA_FIELDS = ["target_key", "monitor_effect_id", "cadence", "next_due_at", + "expires_at", "last_checked_at", "result_hash", "consecutive_no_change", "material_change", + "material_change_generation", "max_no_change_before_replan", "watch_only"] as const; + +function text(value: unknown): string { + if (value === null || value === undefined || value === false || value === 0) return ""; + return stripPythonWhitespace(value === true ? "True" : String(value)); +} + +function timestampOrNull(value: unknown): bigint | null { + return parseTodoTimestampMicros(text(value)); +} + +function timestamp(value: unknown, label: string): bigint { + const result = timestampOrNull(value); + if (result === null) throw new EffectRuntimeRequestError(`${label} must be an ISO timestamp`); + return result; +} + +function schedule(generatedAt: string, cadence: unknown, explicit: unknown = null) { + const result = evaluateSchedulerStateTransition({ + schema_version: SCHEDULER_STATE_TRANSITION_REQUEST_SCHEMA, operation: "monitor_schedule", + generated_at: generatedAt, cadence: cadence ?? null, explicit_next_due_at: explicit ?? null, + }); + if (result.operation !== "monitor_schedule") throw new Error("monitor schedule result mismatch"); + return result; +} + +function counter(value: unknown): number { + const raw = text(value); + if (!/^[+-]?\d+$/.test(raw)) return 0; + const result = Number(raw); + if (!Number.isSafeInteger(result)) throw new EffectRuntimeRequestError("monitor counter exceeds the safe integer range"); + return Math.max(0, result); +} + +function normalizeMetadata(value: unknown): JsonObject { + const raw = value == null ? {} : requireJsonObject(value, "monitor metadata"); + const normalized: JsonObject = {}; + for (const field of MONITOR_METADATA_FIELDS) { + if (!Object.hasOwn(raw, field)) continue; + if (raw[field] === null) normalized[field] = null; + else if (text(raw[field])) normalized[field] = text(raw[field]); + } + if (normalized.cadence != null && schedule("1970-01-01T00:00:00Z", normalized.cadence).cadence_seconds === null) { + throw new EffectRuntimeRequestError("--cadence must look like 30m, 2h, or 1d"); + } + for (const field of ["next_due_at", "expires_at", "last_checked_at"] as const) { + if (normalized[field] != null) timestamp(normalized[field], `--${field.replaceAll("_", "-")}`); + } + for (const field of ["consecutive_no_change", "material_change_generation"] as const) { + if (normalized[field] == null) continue; + const raw = String(normalized[field]); + const label = `--${field.replaceAll("_", "-")}`; + if (!/^[+-]?\d+$/.test(raw)) throw new EffectRuntimeRequestError(`${label} must be an integer`); + if (!Number.isSafeInteger(Number(raw)) || Number(raw) < 0) { + throw new EffectRuntimeRequestError(`${label} must be a non-negative safe integer`); + } + } + if (normalized.material_change != null && !["true", "false"].includes(String(normalized.material_change))) { + throw new EffectRuntimeRequestError("--material-change metadata must be true or false"); + } + if (normalized.watch_only != null && !["true", "false"].includes(String(normalized.watch_only).toLowerCase())) { + throw new EffectRuntimeRequestError("--watch-only metadata must be true or false"); + } + return normalized; +} + +function poll(existing: JsonObject, observation: JsonObject): {metadata: JsonObject; transition: JsonObject} { + if (existing.task_class !== "continuous_monitor") { + throw new EffectRuntimeRequestError("monitor poll observation requires task_class=continuous_monitor"); + } + const material = requireBoolean(observation.material_change, "material_change"); + const resultHash = text(observation.result_hash); + if (!resultHash) throw new EffectRuntimeRequestError("monitor todo writeback requires --result-hash"); + const generatedAt = text(observation.generated_at); + const observedAt = timestamp(generatedAt, "generated_at"); + const existingTarget = text(existing.target_key); + const requestedTarget = text(observation.target_key); + if (requestedTarget && existingTarget && requestedTarget !== existingTarget) { + throw new EffectRuntimeRequestError(`monitor poll target_key resolves to '${existingTarget}', not '${requestedTarget}'`); + } + const target = requestedTarget || existingTarget; + const cadence = text(observation.cadence || existing.cadence); + const due = schedule(generatedAt, cadence, observation.next_due_at).next_due_at; + if (!material && !due) throw new EffectRuntimeRequestError( + "unchanged monitor todo writeback requires --next-due-at or a parseable cadence such as 30m/2h/1d"); + const effectId = text(observation.monitor_effect_id); + const previousEffect = text(existing.monitor_effect_id); + const previousGeneration = counter(existing.material_change_generation); + const previousNoChange = counter(existing.consecutive_no_change); + const replay = Boolean(effectId && effectId === previousEffect); + if (replay) { + const facts = {result_hash: resultHash, material_change: String(material), last_checked_at: generatedAt, + target_key: target, cadence, next_due_at: due || ""}; + const conflicts = Object.entries(facts).filter(([key, expected]) => text(existing[key]) !== expected).map(([key]) => key); + if (conflicts.length) throw new EffectRuntimeRequestError( + `monitor effect identity is already bound to different observation fields: ${conflicts.join(", ")}`); + } else { + const persistedAt = timestampOrNull(existing.last_checked_at); + // Ordering is about observations, not whether a caller supplied a retry ID. + // Preserve same-second unkeyed polls; distinct keyed effects retain strict ordering. + if (persistedAt !== null && (observedAt < persistedAt || + (effectId && previousEffect && observedAt === persistedAt))) { + throw new EffectRuntimeRequestError("monitor observation is older than the persisted monitor effect"); + } + } + const previousHash = text(existing.result_hash); + const advances = !replay && material && resultHash !== previousHash; + const generation = previousGeneration + Number(advances); + const noChange = replay ? previousNoChange : material || (previousHash && previousHash !== resultHash) + ? 0 : previousNoChange + 1; + if (!Number.isSafeInteger(generation) || !Number.isSafeInteger(noChange)) { + throw new EffectRuntimeRequestError("monitor counter exceeds the safe integer range"); + } + const metadata: JsonObject = replay + ? Object.fromEntries(MONITOR_METADATA_FIELDS.filter(key => existing[key] != null).map(key => [key, existing[key]])) + : {last_checked_at: generatedAt, result_hash: resultHash, consecutive_no_change: String(noChange), + material_change: String(material), material_change_generation: String(generation)}; + if (!replay) { + if (effectId) metadata.monitor_effect_id = effectId; + if (target) metadata.target_key = target; + if (cadence) metadata.cadence = cadence; + if (due) metadata.next_due_at = due; + } + return {metadata, transition: {monitor_effect_id: effectId || null, provider_replayed: replay, + result_hash: resultHash, material_change: material, material_change_applied: advances, + material_change_generation: generation, consecutive_no_change: noChange, last_checked_at: generatedAt, + target_key: target || null, cadence: cadence || null, next_due_at: due}}; +} + +export interface MonitorMetadataPlan extends JsonObject { + schema_version: typeof TODO_MONITOR_METADATA_RESULT_SCHEMA; + metadata: JsonObject; + transition: JsonObject | null; +} + +/** Composed in-process by the field planner, or used by the create adapter. */ +export function planMonitorMetadata(value: unknown): MonitorMetadataPlan { + const request = requireJsonObject(value, "monitor metadata request"); + if (request.schema_version !== TODO_MONITOR_METADATA_REQUEST_SCHEMA) { + throw new EffectRuntimeRequestError("monitor metadata request schema mismatch"); + } + const existing = requireJsonObject(request.existing ?? {}, "monitor source"); + const planned = request.observation == null ? null : poll(existing, requireJsonObject(request.observation, "monitor observation")); + if (planned && request.metadata != null) throw new EffectRuntimeRequestError("monitor observation cannot be combined with raw metadata"); + let metadata = normalizeMetadata(planned?.metadata ?? request.metadata); + const nonTarget = Object.entries(metadata).some(([key, v]) => key !== "target_key" && v !== null); + if (nonTarget && (request.role !== "agent" || request.task_class !== "continuous_monitor")) { + throw new EffectRuntimeRequestError("monitor schedule metadata requires --role agent --task-class continuous_monitor"); + } + if (metadata.target_key != null && request.role !== "agent") throw new EffectRuntimeRequestError("target_key requires --role agent"); + if (request.generated_at != null && request.task_class === "continuous_monitor" && metadata.next_due_at == null && metadata.cadence != null) { + const due = schedule(text(request.generated_at), metadata.cadence).next_due_at; + if (due) metadata = {...metadata, next_due_at: due}; + } + if (requireBoolean(request.enforce_boundedness, "enforce_boundedness") && request.task_class === "continuous_monitor") { + const effective = {...existing, ...metadata}; + if (!text(effective.expires_at) && !text(request.resume_when) && text(effective.watch_only).toLowerCase() !== "true") { + throw new EffectRuntimeRequestError("continuous_monitor requires one of: --expires-at, --resume-when, or --watch-only"); + } + } + return {schema_version: TODO_MONITOR_METADATA_RESULT_SCHEMA, metadata, transition: planned?.transition ?? null}; +} diff --git a/loopx/control_plane/work_items/task_lease_acquire.ts b/loopx/control_plane/work_items/task_lease_acquire.ts index 0640e6a261..c0217f2d92 100644 --- a/loopx/control_plane/work_items/task_lease_acquire.ts +++ b/loopx/control_plane/work_items/task_lease_acquire.ts @@ -1,4 +1,5 @@ import { ShadowManagementError, requireShadowPrimaryWriteAllowed } from "../coordination/shadow_management.ts"; +import { parseIsoTimestamp } from "../runtime_timestamp.ts"; import { LegacyCoordinationWriteError, requireLegacyCoordinationPrimaryWriteAllowed } from "../coordination/legacy_writer_fence.ts"; import { createHash } from "node:crypto"; import { readdir, readFile } from "node:fs/promises"; @@ -571,44 +572,6 @@ export function leaseEpoch(lease: LeaseRecord | null): number { return leaseInteger(lease, "lease_epoch") ?? 1; } -export function parseLeaseTimestamp(value: string): Date | null { - const match = /^(\d{4})-(\d{2})-(\d{2})(?:[T ](\d{2}):(\d{2})(?::(\d{2})(?:\.(\d+))?)?(Z|z|[+-]\d{2}(?::?\d{2})?)?)?$/u.exec( - value.trim(), - ); - if (match === null) return null; - const [, yearText, monthText, dayText, hourText, minuteText, secondText, fraction, timezone] = match; - const [year, month, day, hour, minute, second, millisecond] = [ - yearText, - monthText, - dayText, - hourText ?? "0", - minuteText ?? "0", - secondText ?? "0", - (fraction ?? "").slice(0, 3).padEnd(3, "0") || "0", - ].map(Number); - const endOfDay = hour === 24; - if ( - endOfDay && - (minute !== 0 || second !== 0 || (fraction !== undefined && /[1-9]/u.test(fraction))) - ) return null; - const calendarHour = endOfDay ? 0 : hour; - const calendar = new Date(0); - calendar.setUTCHours(calendarHour, minute, second, millisecond); - calendar.setUTCFullYear(year, month - 1, day); - if ( - calendar.getUTCFullYear() !== year || calendar.getUTCMonth() !== month - 1 || - calendar.getUTCDate() !== day || calendar.getUTCHours() !== calendarHour || - calendar.getUTCMinutes() !== minute || calendar.getUTCSeconds() !== second || - calendar.getUTCMilliseconds() !== millisecond - ) return null; - if (hourText === undefined) return calendar; - let text = value.trim().replace(" ", "T").replace(/z$/u, "Z"); - if (fraction !== undefined) text = text.replace(`.${fraction}`, `.${fraction.slice(0, 3)}`); - if (timezone === undefined) text += "Z"; - else text = text.replace(/([+-]\d{2})$/u, "$1:00"); - const parsed = new Date(text); - return Number.isNaN(parsed.valueOf()) ? null : parsed; -} export function leaseIsActive(lease: LeaseRecord | null, at: Date): boolean { if ( @@ -624,7 +587,7 @@ export function leaseIsActive(lease: LeaseRecord | null, at: Date): boolean { { expires_at: lease.expires_at ?? null }, ); } - const expiresAt = parseLeaseTimestamp(lease.expires_at); + const expiresAt = parseIsoTimestamp(lease.expires_at); if (expiresAt === null) { throw new TaskLeaseAcquireError( "active lease expires_at must be a valid timestamp", diff --git a/loopx/todos.py b/loopx/todos.py index 6927d5b4f1..e353174dfe 100644 --- a/loopx/todos.py +++ b/loopx/todos.py @@ -797,11 +797,8 @@ def add_goal_todo( updated_at = now_local() normalized_monitor_metadata = todo_monitor_metadata.require_monitor_metadata_scope( monitor_metadata=monitor_metadata, role=role, task_class=task_class, - generated_at=updated_at, - ) - todo_monitor_metadata.require_continuous_monitor_boundedness( - task_class=task_class, resume_when=normalized_resume_when, - monitor_metadata=normalized_monitor_metadata, + generated_at=updated_at, resume_when=normalized_resume_when, + enforce_boundedness=True, ) canonical_create = create_canonical_todo_if_promoted( registry_path=registry_path, @@ -1222,12 +1219,7 @@ def update_goal_todo( raise ValueError(f"todo_id {normalized_todo_id!r} was not found in active user or agent todos") existing_role, _section, _start, _end, existing_block = existing_block_match target_role = role or existing_role - monitor_metadata_input, monitor_poll_transition = ( - todo_monitor_metadata.resolve_monitor_metadata_input( - existing=existing_block, - monitor_metadata=monitor_metadata, - ) - ) + monitor_intent = todo_monitor_metadata.monitor_metadata_intent(monitor_metadata) authority_todo = dict(existing_block) authority_todo["role"] = target_role authority_action = todo_update_authority_action( @@ -1247,7 +1239,7 @@ def update_goal_todo( clear_global_gate, unblocks_todo_id, successor_todo_ids, resume_when, clear_resume_when, no_followup, ), - monitor_metadata=monitor_metadata_input, + monitor_metadata=monitor_intent["observation"] or monitor_intent["metadata"], ) mutation_authority = authorize_todo_lifecycle_mutation( registry_path=registry_path, @@ -1336,14 +1328,6 @@ def update_goal_todo( task_class=target_task_class, ) ) - normalized_monitor_metadata = todo_monitor_metadata.validate_monitor_metadata_update( - monitor_metadata=monitor_metadata_input, - existing=existing_block, - role=target_role, - task_class=target_task_class, generated_at=updated_at, - resume_when=effective_resume_when, - enforce_boundedness=enforce_monitor_boundedness, - ) update_result = apply_todo_update_to_lines( lines, todo_id=todo_id, @@ -1386,7 +1370,11 @@ def update_goal_todo( completion_metadata_updates_override=( completion_metadata_updates_override ), - monitor_metadata=normalized_monitor_metadata, + monitor_context={ + **monitor_intent, "role": target_role, "task_class": target_task_class, + "resume_when": effective_resume_when, + "enforce_boundedness": enforce_monitor_boundedness, + }, clear_claim=clear_claim, claim_only=claim_only, updated_at=updated_at, @@ -1422,8 +1410,6 @@ def update_goal_todo( payload["parent_successor_advisory"] = parent_successor_advisory if external_wait_transition is not None: payload["external_wait_transition"] = external_wait_transition - if monitor_poll_transition is not None: - payload["monitor_poll_transition"] = monitor_poll_transition payload = _attach_todo_write_correctness_dry_run_packet( payload, goal_id=goal_id, From f93768f47f3d72eac44fee03e7cdecd396fdcfc8 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 9 Sep 2026 23:56:20 +0800 Subject: [PATCH 2/3] test(monitor): qualify state parity and document migration boundaries Signed-off-by: huangruiteng --- ...shared-goal-authority-state-provider-v0.md | 7 + ...-goal-authority-state-provider-v0.zh-CN.md | 5 + .../typescript-control-plane-migration-v0.md | 18 +++ ...script-control-plane-migration-v0.zh-CN.md | 14 ++ docs/project-agent-todo-contract.md | 23 +++ .../monitor-todo-policy-seam-smoke.py | 3 - .../todo-readmodel-boundary-smoke.py | 6 +- ...sue_fix_grouped_monitor_materialization.py | 35 +++- .../control_plane/test_monitor_state_owner.py | 98 ++++++++++++ .../control_plane_ts/monitor_metadata.test.ts | 150 ++++++++++++++++++ tsconfig.control-plane.json | 1 + 11 files changed, 353 insertions(+), 7 deletions(-) create mode 100644 tests/control_plane/test_monitor_state_owner.py create mode 100644 tests/control_plane_ts/monitor_metadata.test.ts diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md index 3c0574728f..be3a160202 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -2524,6 +2524,13 @@ commands fail closed; they do not fall back to the old writer. #### Refactoring roadmap overview +The Monitor state owner now lives in TS and is composed by the legacy update +field plan. This removes Python poll/generation and metadata rules, but the +legacy writer still holds the lock and commits the result. The typed plan is +not an authority receipt; monitor/successor atomicity, native metadata update, +provider defaults and D1–D3 remain separate, unfinished gates. Permanent +Markdown projection remains part of the target architecture. + Todo authoring scope and terminal successors now share the TS resolved-binding invariant. Only explicit `global_gate` can widen blocking to all registered agents; `goal_bound` grants no global-gate semantics. This consolidates T1 diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md index cd71ce751c..5a44fad64f 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md @@ -2002,6 +2002,11 @@ backend、实时双向同步或按命令拆开的权威；晋升后不支持的 #### 重构主线总览 +Monitor 状态 owner 现位于 TS，并由 legacy update field plan 组合调用；删除 Python +poll/generation 与 metadata 规则，但持锁及结果提交仍由 legacy writer 负责。 +Typed plan 不是 authority receipt；Monitor/successor 原子性、原生 metadata update、 +provider 默认值及 D1–D3 仍是独立、未完成的门禁。永久 Markdown 投影仍属于终态架构。 + Todo authoring scope 已与 terminal successor 共用 TS 最终绑定不变量；仅显式 `global_gate` 可以扩大阻塞到全部注册 agent，`goal_bound` 不授予全局 gate 语义。 这是 T1 准入规则收拢；不扩张 native update 字段权限、不改变 provider/profile 默认值， diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 647f4e9b94..eeb2465cbd 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -16,6 +16,24 @@ ## Current implementation checkpoint +Monitor metadata authoring and poll transitions now share `todos/monitor_metadata.ts`. +Public update composes that owner inside its existing field-plan request; cadence +calculation stays in-process instead of making two additional scheduler RPCs. +The Python observation/replay/counter/scope/boundedness rules are retired. Create +and the low-level Markdown add codec retain a metadata-plan adapter; this is not +the complete T1 transaction or T2 atomic monitor-plus-successor commit. + +Intentional corrections: older observations cannot rewind state merely because +either effect ID is absent; issue-fix grouped membership updates use the locked +observation path and advance generation when a material result hash changes. +New counters reject negative or unsafe integers. ISO dates are calendar-checked; +the codec retains Python compact/week-date forms, offset seconds and microsecond +ordering without rewriting history. Lifecycle/ownership admission now precedes poll +diagnostics, so an unauthorized request cannot use malformed metadata to avoid +its authority rejection. Exact replay, same-second unkeyed polls, explicit +clears and legacy boundedness exemptions remain. The plan grants no permission, +receipt or promotion; native update still owns only text/note. + Public Todo add/update now resolve role, continuation binding, gate scope and deferred-condition requirements through `todos/authoring_scope.ts`. Python's `write_policy.py` and duplicated scope selection in `todos.py` are retired; diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 595aaf47d7..98d5cd7080 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -15,6 +15,20 @@ ## 当前实现检查点 +Monitor metadata authoring 与 poll transition 现共用 `todos/monitor_metadata.ts`。 +公开 update 在已有 field-plan 请求内组合该 owner；cadence 在进程内计算，不再额外 +调用两次 scheduler RPC。删除 Python 的 observation/replay/counter/scope/boundedness +规则。Create 与低层 Markdown add codec 仍保留 metadata-plan adapter；这不是完整 +T1 事务，也不是 T2 的 Monitor 与 successor 原子提交。 + +有意修正：不再因任一 effect ID 缺失而允许旧 observation 倒退状态；issue-fix 分组 +成员更新使用持锁 observation 路径，在 material result hash 改变时递增 generation。 +新计数拒绝负数及不安全整数。ISO 日期进行日历校验，codec 保留 Python 的紧凑日期、 +周日期、时区偏移秒数及微秒排序，不改写历史。 +Lifecycle/ownership 准入现在先于 poll 诊断，未授权请求不能靠非法 metadata 回避 +权限拒绝。精确 replay、同秒无 ID 轮询、显式清空及 legacy boundedness 豁免保持。 +Plan 不授予权限、receipt 或 promotion；native update 仍只拥有 text/note。 + 公开 Todo add/update 现通过 `todos/authoring_scope.ts` 统一解析角色、continuation 绑定、gate 作用域与 deferred 条件要求。删除 Python `write_policy.py` 及 `todos.py` 重复的 scope 选择；Markdown codec 只保留早期 class 检查的适配调用。已物化的 terminal diff --git a/docs/project-agent-todo-contract.md b/docs/project-agent-todo-contract.md index 2f37b8851a..9fd3435d54 100644 --- a/docs/project-agent-todo-contract.md +++ b/docs/project-agent-todo-contract.md @@ -102,6 +102,29 @@ eligible time, `--cadence` is the retry interval, `--monitor-target-key` is the stable idempotency key, and optional `--expires-at` is the hard stop after which the monitor must not catch up. +Monitor observations are reduced against the Todo under its existing writer lock. +Callers report a result hash and material-change fact; they must not independently +increment counters. A material observation with a different result hash increments +`material_change_generation`; repeating the same hash does not. An unchanged +same-hash poll increments `consecutive_no_change`; material change or a changed hash +resets that count. Issue-fix grouped membership updates use this same path. +New grouped monitors default to watch-only; subsequent observations preserve +the existing expiration/watch policy rather than silently re-enabling watch-only. + +An exact `monitor_effect_id` replay keeps the committed counters. Reusing the ID +with different observation fields fails. Older timestamps fail even without an +effect ID; same-second unkeyed polls remain allowed, while distinct keyed effects +retain strict ordering. Ordering preserves microseconds. Newly written +`material_change_generation` and `consecutive_no_change` values must be +non-negative safe integers. Use ISO timestamps +(for example `2030-01-01T12:00:00.000001+00:00`); invalid calendar dates are +rejected. Existing compact/week-date and timezone-offset-second spellings remain readable. +Existing malformed historical timestamps do not prove an ordering fence. + +These rules do not make a Monitor executable delivery work, grant claim/lease +authority, or make Monitor and successor writes atomic. A planning result is +not a durable receipt; provider promotion remains explicitly gated. + Terminology: a `goal_id` is the LoopX control-plane boundary: registry entry, active-state file, quota lane, status projection, and run-history stream. A `todo_id` is a structured work item inside that goal. LoopX does not diff --git a/examples/control_plane/monitor-todo-policy-seam-smoke.py b/examples/control_plane/monitor-todo-policy-seam-smoke.py index bc2e0a61bc..27668fe6ee 100644 --- a/examples/control_plane/monitor-todo-policy-seam-smoke.py +++ b/examples/control_plane/monitor-todo-policy-seam-smoke.py @@ -19,7 +19,6 @@ monitor_todo_is_expired, monitor_todo_missing_schedule, monitor_todo_next_due_at, - parse_monitor_counter, ) from loopx.status import ( # noqa: E402 todo_item_is_due_monitor, @@ -92,8 +91,6 @@ def main() -> int: assert_policy_matches_wrappers(cadence_only, due=False, expired=False) assert monitor_todo_missing_schedule(cadence_only, now=NOW) is True, cadence_only assert monitor_todo_next_due_at({"next_due_at": "2026-01-01T00:00:00"}) == NOW - assert parse_monitor_counter("3") == 3 - assert parse_monitor_counter("not-a-number") == 0 assert monitor_cadence_delta("2h").total_seconds() == 7200 cadence_due_at = monitor_next_due_at( generated_at="2026-01-01T00:00:00+00:00", diff --git a/examples/control_plane/todo-readmodel-boundary-smoke.py b/examples/control_plane/todo-readmodel-boundary-smoke.py index 8020c0d63f..54f6e0d281 100644 --- a/examples/control_plane/todo-readmodel-boundary-smoke.py +++ b/examples/control_plane/todo-readmodel-boundary-smoke.py @@ -108,7 +108,6 @@ def assert_status_compatibility_boundary() -> None: assert status_module.compact_operator_gate_resume_contract is run_compaction_read_model.compact_operator_gate_resume_contract assert status_module.compact_controller_readiness is run_compaction_read_model.compact_controller_readiness assert status_module.parse_timestamp is runtime_time_read_model.parse_timestamp - assert monitor_metadata_read_model.parse_timestamp is runtime_time_read_model.parse_timestamp assert scheduler_time_read_model.parse_timestamp is runtime_time_read_model.parse_timestamp assert status_cache_read_model.parse_timestamp is runtime_time_read_model.parse_timestamp assert evidence_log_read_model.parse_timestamp is runtime_time_read_model.parse_timestamp @@ -158,8 +157,9 @@ def assert_wrapper_parity() -> None: assert stripped is not None assert stripped.isoformat() == "2026-01-01T00:00:00+00:00", stripped assert status_module.parse_timestamp("not-a-time") is None - assert monitor_metadata_read_model.normalize_monitor_metadata( - {"next_due_at": "2026-01-01T00:00:00Z"} + assert monitor_metadata_read_model.require_monitor_metadata_scope( + monitor_metadata={"next_due_at": "2026-01-01T00:00:00Z"}, + role="agent", task_class="continuous_monitor", ) == {"next_due_at": "2026-01-01T00:00:00Z"} diff --git a/tests/capabilities/test_issue_fix_grouped_monitor_materialization.py b/tests/capabilities/test_issue_fix_grouped_monitor_materialization.py index 070cb667ad..9dbf1b24ed 100644 --- a/tests/capabilities/test_issue_fix_grouped_monitor_materialization.py +++ b/tests/capabilities/test_issue_fix_grouped_monitor_materialization.py @@ -27,7 +27,8 @@ from loopx.domain_packs.issue_fix import ( upsert_issue_fix_pr_lifecycle_ledger_jsonl, ) -from loopx.todos import list_goal_todos +from loopx.todos import list_goal_todos, update_goal_todo +from loopx.control_plane.todos.resume_condition import evaluate_todo_resume_conditions GOAL_ID = "issue-fix-monitor-goal" AGENT_ID = "issue-fix-worker" @@ -194,6 +195,38 @@ def test_grouped_monitor_materialization_is_one_per_bucket_and_retires_empty_buc assert "no_followup=true" not in reopened_state +def test_group_membership_change_advances_monitor_generation(tmp_path: Path) -> None: + project, state, registry = _fixture(tmp_path) + ledger = tmp_path / "pr-lifecycle.jsonl" + upsert_issue_fix_pr_lifecycle_ledger_jsonl(ledger, _packet(101)) + arguments = dict(registry_path=registry, goal_id=GOAL_ID, project=project, + ledger_path=ledger, claimed_by=AGENT_ID, cadence="30m") + materialize_issue_fix_grouped_monitors(**arguments, generated_at="2030-01-01T01:00:00Z") + before = _monitor_todos(registry, project)[0] + baseline = int(before.get("material_change_generation") or 0) + waiting = {"todo_id": "todo_waiting", "role": "agent", "status": "open", + "task_class": "advancement_task", "resume_when": f"monitor_changed:{before['todo_id']}", + "resume_monitor_generation": baseline} + assert evaluate_todo_resume_conditions([waiting], source_items=[before])["todo_waiting"]["satisfied"] is False + # Observations must preserve an explicitly bounded existing watch policy, + # rather than resetting it to the capability's initial create default. + update_goal_todo(registry_path=registry, goal_id=GOAL_ID, todo_id=before["todo_id"], + agent_id=AGENT_ID, project=project, + monitor_metadata={"watch_only": "false", "expires_at": "2031-01-01T00:00:00Z"}) + upsert_issue_fix_pr_lifecycle_ledger_jsonl(ledger, _packet(102)) + materialize_issue_fix_grouped_monitors(**arguments, generated_at="2030-01-01T02:00:00Z") + after = _monitor_todos(registry, project)[0] + assert after["todo_id"] == before["todo_id"] + assert after["result_hash"] != before["result_hash"] + assert after["material_change_generation"] == baseline + 1 + assert str(after["watch_only"]).lower() == "false" + assert after["expires_at"] == "2031-01-01T00:00:00Z" + assert evaluate_todo_resume_conditions([waiting], source_items=[after])["todo_waiting"]["satisfied"] is True + unchanged = state.read_bytes() + materialize_issue_fix_grouped_monitors(**arguments, generated_at="2030-01-01T03:00:00Z") + assert state.read_bytes() == unchanged + + def test_grouped_monitors_are_isolated_per_repository(tmp_path: Path) -> None: project, state, registry = _fixture(tmp_path) ledger = tmp_path / "pr-lifecycle.jsonl" diff --git a/tests/control_plane/test_monitor_state_owner.py b/tests/control_plane/test_monitor_state_owner.py new file mode 100644 index 0000000000..7899148b1f --- /dev/null +++ b/tests/control_plane/test_monitor_state_owner.py @@ -0,0 +1,98 @@ +"""Public writer regressions; fixtures never touch an installed Goal.""" +from pathlib import Path + +import pytest + +from loopx.control_plane.todos.monitor_metadata import MonitorPollObservation +from loopx.todos import list_goal_todos, update_goal_todo +from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result +from loopx.control_plane.runtime.time import parse_timestamp +from tests.control_plane.test_monitor_followthrough_contract import ( + AGENT_ID, GOAL_ID, _add_monitor, _write_fixture, +) + + +def _poll(registry: Path, todo_id: str, at: str, result: str, **kwargs): + return update_goal_todo( + registry_path=registry, goal_id=GOAL_ID, todo_id=todo_id, + role="agent", agent_id=AGENT_ID, + monitor_metadata=MonitorPollObservation( + generated_at=at, result_hash=result, material_change=True, **kwargs, + ), + ) + + +@pytest.mark.parametrize("effect_id", [None, "effect-two"]) +def test_older_monitor_observation_cannot_rewind_state(tmp_path, effect_id): + registry, _, state = _write_fixture(tmp_path) + todo = _add_monitor(registry, text="Observe public fixture", target_key="fixture") + _poll(registry, todo["todo_id"], "2030-01-01T02:00:00Z", "new") + before = state.read_bytes() + with pytest.raises(ValueError, match="older"): + _poll(registry, todo["todo_id"], "2030-01-01T01:00:00Z", "old", + monitor_effect_id=effect_id) + assert state.read_bytes() == before + + +def test_invalid_observation_time_is_rejected_even_with_explicit_schedule(tmp_path): + registry, _, state = _write_fixture(tmp_path) + todo = _add_monitor(registry, text="Observe public fixture", target_key="fixture") + before = state.read_bytes() + with pytest.raises(ValueError, match="timestamp"): + _poll(registry, todo["todo_id"], "not-a-time", "new", + next_due_at="2030-01-01T03:00:00Z") + assert state.read_bytes() == before + + +def test_authority_rejection_precedes_invalid_poll_diagnostics(tmp_path): + registry, _, state = _write_fixture(tmp_path) + todo = _add_monitor(registry, text="Observe public fixture", target_key="fixture") + before = state.read_bytes() + with pytest.raises(ValueError, match="claimed_by"): + update_goal_todo(registry_path=registry, goal_id=GOAL_ID, todo_id=todo["todo_id"], + agent_id="codex-main-control", + monitor_metadata=MonitorPollObservation( + generated_at="invalid", result_hash="", material_change=False)) + assert state.read_bytes() == before + + +def test_monitor_effect_replay_and_generation_are_locked_public_writer_semantics(tmp_path): + registry, _, state = _write_fixture(tmp_path) + todo = _add_monitor(registry, text="Observe public fixture", target_key="fixture") + first = _poll(registry, todo["todo_id"], "2030-01-01T02:00:00Z", "new", + monitor_effect_id="effect-one") + assert first["monitor_poll_transition"]["material_change_generation"] == 1 + before = state.read_bytes() + replay = _poll(registry, todo["todo_id"], "2030-01-01T02:00:00Z", "new", + monitor_effect_id="effect-one") + assert replay["monitor_poll_transition"]["provider_replayed"] is True + assert state.read_bytes() == before + with pytest.raises(ValueError, match="different observation"): + _poll(registry, todo["todo_id"], "2030-01-01T02:00:00Z", "other", + monitor_effect_id="effect-one") + assert state.read_bytes() == before + readback = list_goal_todos(registry_path=registry, goal_id=GOAL_ID, role="agent") + assert readback["todos"][0]["material_change_generation"] == 1 + + +@pytest.mark.parametrize("value", [ + "1970-01-01", "19700101", "1970-W01-4", "1970W014", "1970-W01", "1970W01", + "19700101T00", "1970-01-01X0000", "1970-01-01 00:00:00", "1970-01-01T00:00z", + "2030-01-01T12:34:56.123456+08:00", "20300101T123456,123456+0800", + "1970-01-01T01:00:00+00:59:59.999999", "1970-01-01T00.1", + "1970-02-30", "2021-W53", "1970-01-01T24:00:00", "1970-01-01T01:00+24:00", + "1970-01-01T00:0000", "1970-01-01T0000:00", "tomorrow", "2030", +]) +def test_monitor_timestamp_input_matches_retained_python_iso_codec(value): + request = { + "schema_version": "loopx_todo_monitor_metadata_request_v0", + "role": "agent", "task_class": "continuous_monitor", "enforce_boundedness": True, + "metadata": {"expires_at": value}, + } + # The pre-migration stdlib codec is an independent input-compatibility oracle. + if parse_timestamp(value) is None: + with pytest.raises(EffectRuntimeRejected, match="timestamp"): + effect_runtime_result("todo.monitor_metadata.plan", request) + else: + result = effect_runtime_result("todo.monitor_metadata.plan", request) + assert result["metadata"]["expires_at"] == value diff --git a/tests/control_plane_ts/monitor_metadata.test.ts b/tests/control_plane_ts/monitor_metadata.test.ts new file mode 100644 index 0000000000..d86dc26ada --- /dev/null +++ b/tests/control_plane_ts/monitor_metadata.test.ts @@ -0,0 +1,150 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { planMonitorMetadata, TODO_MONITOR_METADATA_REQUEST_SCHEMA as SCHEMA } from "../../loopx/control_plane/todos/monitor_metadata.ts"; +import { planTodoFieldUpdate, TODO_FIELD_UPDATE_REQUEST_SCHEMA } from "../../loopx/control_plane/todos/field_update.ts"; +import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; +import { productionScaleCoordinationFixture } from "./production_scale_coordination_fixture.ts"; +import { parseTodoTimestampMicros } from "../../loopx/control_plane/runtime_timestamp.ts"; + +const existing = {todo_id: "todo_monitor", task_class: "continuous_monitor", status: "open", + target_key: "fixture", cadence: "1h", watch_only: "true", result_hash: "before", + material_change_generation: "7", consecutive_no_change: "3"}; +function request(extra: JsonObject = {}): JsonObject { + return {schema_version: SCHEMA, existing, role: "agent", task_class: "continuous_monitor", + generated_at: "2030-01-01T00:00:00Z", enforce_boundedness: true, ...extra}; +} +function observation(extra: JsonObject = {}): JsonObject { + return {generated_at: "2030-01-01T01:00:00Z", result_hash: "after", material_change: true, + monitor_effect_id: "effect-a", ...extra}; +} + +test("generation and no-change count follow result semantics, not mere poll count", () => { + for (const [material, hash, generation, count] of [ + [true, "after", 8, 0], [true, "before", 7, 0], + [false, "after", 7, 0], [false, "before", 7, 4], + ] as const) { + const result = planMonitorMetadata(request({observation: observation({material_change: material, result_hash: hash})})); + assert.equal(result.metadata.material_change_generation, String(generation)); + assert.equal(result.metadata.consecutive_no_change, String(count)); + assert.equal(result.transition?.material_change_applied, generation === 8); + assert.equal(result.metadata.next_due_at, "2030-01-01T02:00:00Z"); + } +}); + +test("exact retry reuses counters; different effect intent cannot reuse its ID", () => { + const original = structuredClone(existing); + const first = planMonitorMetadata(request({observation: observation()})); + assert.deepEqual(existing, original); + const persisted = {...existing, ...first.metadata}; + const replay = planMonitorMetadata(request({existing: persisted, observation: observation()})); + assert.equal(replay.transition?.provider_replayed, true); + assert.equal(replay.transition?.material_change_applied, false); + assert.deepEqual(replay.metadata, Object.fromEntries(Object.entries(persisted).filter(([key]) => + !["todo_id", "task_class", "status"].includes(key)))); + for (const conflicting of [{result_hash: "other"}, {material_change: false}, {cadence: "2h"}, + {generated_at: "2030-01-01T02:00:00Z"}]) { + assert.throws(() => planMonitorMetadata(request({existing: persisted, observation: observation(conflicting)})), /different observation/); + } +}); + +test("out-of-order rejection does not depend on effect IDs", () => { + for (const previousId of [null, "old-effect"]) for (const nextId of [null, "next-effect"]) { + assert.throws(() => planMonitorMetadata(request({ + existing: {...existing, last_checked_at: "2030-01-01T02:00:00Z", monitor_effect_id: previousId}, + observation: observation({monitor_effect_id: nextId}), + })), /older/); + } + assert.doesNotThrow(() => planMonitorMetadata(request({ + existing: {...existing, last_checked_at: "2030-01-01T01:00:00Z"}, + observation: observation({monitor_effect_id: null}), + }))); +}); + +test("create/edit scope, boundedness and explicit clearing share one owner", () => { + assert.deepEqual(planMonitorMetadata(request({existing: {}, metadata: {cadence: "30m", watch_only: "true"}})).metadata, + {cadence: "30m", watch_only: "true", next_due_at: "2030-01-01T00:30:00Z"}); + assert.throws(() => planMonitorMetadata(request({metadata: {watch_only: null}})), /requires one of/); + assert.doesNotThrow(() => planMonitorMetadata(request({metadata: {watch_only: null}, resume_when: "todo_done:todo_dependency"}))); + assert.doesNotThrow(() => planMonitorMetadata(request({metadata: {watch_only: null}, enforce_boundedness: false}))); + assert.throws(() => planMonitorMetadata(request({role: "user", metadata: {cadence: "1h"}})), /schedule metadata/); + assert.throws(() => planMonitorMetadata(request({role: "user", metadata: {target_key: "fixture"}})), /target_key/); + assert.throws(() => planMonitorMetadata(request({metadata: {cadence: "never"}})), /cadence/); + for (const field of ["consecutive_no_change", "material_change_generation"]) { + for (const value of ["-1", "9007199254740993", "1.5"]) { + assert.throws(() => planMonitorMetadata(request({metadata: {[field]: value}})), /integer/); + } + } +}); + +test("public update field plan composes the observation once and keeps completion independent", () => { + const result = planTodoFieldUpdate({schema_version: TODO_FIELD_UPDATE_REQUEST_SCHEMA, + todo: existing, updated_at: "2030-01-01T01:00:00Z", intent: {note: "Checked"}, + monitor_context: {role: "agent", task_class: "continuous_monitor", enforce_boundedness: true, + observation: observation()}, + }); + assert.equal(result.metadata_updates.material_change_generation, "8"); + assert.equal(result.metadata_updates.note, "Checked"); + assert.equal(result.target_status, "open"); + assert.equal((result.monitor_poll_transition as JsonObject).provider_replayed, false); + assert.equal(result.metadata_updates.claimed_by, undefined); +}); + +test("poll rejects invalid observation shapes without mutating its input", () => { + for (const invalid of [{material_change: "false"}, {generated_at: "invalid", next_due_at: "2030-01-01T03:00:00Z"}, + {result_hash: ""}, {target_key: "different"}]) { + const source = request({observation: observation(invalid)}); + const before = structuredClone(source); + assert.throws(() => planMonitorMetadata(source)); + assert.deepEqual(source, before); + } +}); + +test("timestamp codec rejects rollover dates and ordering retains microseconds", () => { + for (const invalid of ["2030-02-30T00:00:00Z", "2030-01-01T24:00:00Z", "2030"]) { + assert.throws(() => planMonitorMetadata(request({metadata: {expires_at: invalid}})), /timestamp/); + } + assert.throws(() => planMonitorMetadata(request({ + existing: {...existing, last_checked_at: "2030-01-01T01:00:00.000002Z"}, + observation: observation({generated_at: "2030-01-01T01:00:00.000001Z", monitor_effect_id: null}), + })), /older/); + assert.doesNotThrow(() => planMonitorMetadata(request({ + existing: {...existing, last_checked_at: "invalid historical date"}, observation: observation(), + }))); +}); + +test("production-scale snapshot remains immutable while each monitor gets an isolated plan", () => { + const fixture = productionScaleCoordinationFixture("goal-fixture"); + const before = structuredClone(fixture); + const todos = fixture.projection.todos as JsonObject[]; + const monitors = todos.filter(todo => todo.task_class === "continuous_monitor"); + assert.equal(todos.length, 464); + assert.equal((fixture.projection.leases as unknown[]).length, 64); + assert.equal(monitors.length, 63); + for (const todo of monitors) { + // A pure plan does not grant permission to mutate archived/completed or + // lease-bearing work. Preserve the complete fixture and test the rule only. + const result = planMonitorMetadata(request({existing: todo, enforce_boundedness: false, + observation: observation({target_key: null, cadence: "1h"}), + })); + assert.equal(result.metadata.material_change_generation, "1"); + assert.equal(result.metadata.consecutive_no_change, "0"); + assert.equal(result.metadata.claimed_by, undefined); + assert.equal(result.metadata.status, undefined); + } + assert.deepEqual(fixture, before); +}); + +test("Todo timestamp codec retains Python ISO compatibility, timezone seconds and exact microseconds", () => { + for (const value of ["1970-01-01", "19700101", "1970-W01-4", "1970W014", + "19700101T00", "1970-01-01X0000", "1970-01-01 00:00:00", "1970-01-01T00:00z"]) { + assert.equal(parseTodoTimestampMicros(value), 0n, value); + } + for (const value of ["1970-01-01T00:00:00.000001Z", "19700101T000000,000001", + "1970-01-01T01:00:00+00:59:59.999999"]) { + assert.equal(parseTodoTimestampMicros(value), 1n, value); + } + for (const value of ["1970-02-30", "2021-W53", "1970-01-01T24:00:00", "1970-01-01T01:00+24:00", + "1970-01-01T00:0000", "1970-01-01T0000:00", "tomorrow", "2030"]) { + assert.equal(parseTodoTimestampMicros(value), null, value); + } +}); diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index 22c7023a1b..3c162c0263 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -66,6 +66,7 @@ "loopx/control_plane/work_items/task_lease_acquire_cli.ts", "examples/nokv-authority-store/live-qualification.ts", "tests/control_plane_ts/effect_program.test.ts", + "tests/control_plane_ts/monitor_metadata.test.ts", "tests/control_plane_ts/effect_runtime_errors.test.ts", "tests/control_plane_ts/post_writeback_hook_transaction.test.ts", "tests/control_plane_ts/interaction_contract.test.ts", From af151c965b37a034f2f7fa4a902b471b4ea31b4b Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 10 Sep 2026 16:00:54 +0800 Subject: [PATCH 3/3] fix(monitor): preserve legacy timestamp separator rejection Signed-off-by: huangruiteng --- loopx/control_plane/runtime_timestamp.ts | 6 +++-- .../control_plane/test_monitor_state_owner.py | 26 +++++++++++++++++++ .../control_plane_ts/monitor_metadata.test.ts | 12 +++++++++ 3 files changed, 42 insertions(+), 2 deletions(-) diff --git a/loopx/control_plane/runtime_timestamp.ts b/loopx/control_plane/runtime_timestamp.ts index 80b7e3668d..668e29eb28 100644 --- a/loopx/control_plane/runtime_timestamp.ts +++ b/loopx/control_plane/runtime_timestamp.ts @@ -42,7 +42,9 @@ export function parseIsoTimestamp(value: string): Date | null { * It keeps microseconds and offset seconds, which a JS Date cannot represent. * Missing timezone means UTC, matching the existing Python runtime codec. */ export function parseTodoTimestampMicros(value: string): bigint | null { - const match = /^(\d{4}-\d{2}-\d{2}|\d{8}|\d{4}-W\d{2}(?:-[1-7])?|\d{4}W\d{2}[1-7]?)(?:[\s\S](.+))?$/u.exec(value); + // The legacy wrapper replaces Z/z with +00:00 before fromisoformat, so + // these letters are timezone suffixes, never date/time separators. + const match = /^(\d{4}-\d{2}-\d{2}|\d{8}|\d{4}-W\d{2}(?:-[1-7])?|\d{4}W\d{2}[1-7]?)(?:[^Zz](.+))?$/u.exec(value); if (!match) return null; const [, date, time] = match; let calendar: Date | null; @@ -64,7 +66,7 @@ export function parseTodoTimestampMicros(value: string): bigint | null { if (time === undefined) return BigInt(calendar.valueOf()) * 1000n; const parts = /^(.*?)(Z|z|[+-].*)?$/.exec(time)!; function clock(raw: string, offset: boolean): bigint | null { - const parsed = /^(\d{2})(?:(:?)?(\d{2})(?:\2(\d{2}))?)?(?:[.,](\d+))?$/.exec(raw); + const parsed = /^(\d{2})(?:(:?)(\d{2})(?:\2(\d{2}))?)?(?:[.,](\d+))?$/.exec(raw); if (!parsed) return null; const hour = Number(parsed[1]), minute = Number(parsed[3] ?? 0), second = Number(parsed[4] ?? 0); if (!offset && (hour > 23 || minute > 59 || second > 59)) return null; diff --git a/tests/control_plane/test_monitor_state_owner.py b/tests/control_plane/test_monitor_state_owner.py index 7899148b1f..0432a6b510 100644 --- a/tests/control_plane/test_monitor_state_owner.py +++ b/tests/control_plane/test_monitor_state_owner.py @@ -96,3 +96,29 @@ def test_monitor_timestamp_input_matches_retained_python_iso_codec(value): else: result = effect_runtime_result("todo.monitor_metadata.plan", request) assert result["metadata"]["expires_at"] == value + + +@pytest.mark.parametrize("date", ["1970-01-01", "19700101", "1970-W01-4"]) +@pytest.mark.parametrize("separator", ["Z", "z"]) +def test_public_writer_rejects_timezone_letter_as_date_separator(tmp_path, date, separator): + registry, _, state = _write_fixture(tmp_path) + todo = _add_monitor(registry, text="Observe public fixture", target_key="fixture") + value = f"{date}{separator}00:00" + assert parse_timestamp(value) is None # Independent legacy input contract. + before = state.read_bytes() + with pytest.raises(ValueError, match="expires-at must be an ISO timestamp"): + update_goal_todo(registry_path=registry, goal_id=GOAL_ID, todo_id=todo["todo_id"], + role="agent", agent_id=AGENT_ID, monitor_metadata={"expires_at": value}) + assert state.read_bytes() == before + + +@pytest.mark.parametrize("suffix", ["Z", "z"]) +def test_public_writer_keeps_terminal_timezone_letter(tmp_path, suffix): + registry, _, _state = _write_fixture(tmp_path) + todo = _add_monitor(registry, text="Observe public fixture", target_key="fixture") + value = f"2030-01-01T00:00{suffix}" + assert parse_timestamp(value) is not None + update_goal_todo(registry_path=registry, goal_id=GOAL_ID, todo_id=todo["todo_id"], + role="agent", agent_id=AGENT_ID, monitor_metadata={"expires_at": value}) + readback = list_goal_todos(registry_path=registry, goal_id=GOAL_ID, role="agent") + assert readback["todos"][0]["expires_at"] == value diff --git a/tests/control_plane_ts/monitor_metadata.test.ts b/tests/control_plane_ts/monitor_metadata.test.ts index d86dc26ada..9852ad458a 100644 --- a/tests/control_plane_ts/monitor_metadata.test.ts +++ b/tests/control_plane_ts/monitor_metadata.test.ts @@ -112,6 +112,18 @@ test("timestamp codec rejects rollover dates and ordering retains microseconds", }))); }); +test("timestamp timezone letters are suffixes, not date separators", () => { + for (const date of ["1970-01-01", "19700101", "1970-W01-4"]) { + for (const letter of ["Z", "z"]) { + assert.equal(parseTodoTimestampMicros(`${date}${letter}00:00`), null); + assert.equal(parseTodoTimestampMicros(`${date}T00:00${letter}`), 0n); + } + } + for (const time of ["00", "0000", "00:00", "000000", "00:00:00"]) { + assert.equal(parseTodoTimestampMicros(`1970-01-01T${time}Z`), 0n); + } +}); + test("production-scale snapshot remains immutable while each monitor gets an isolated plan", () => { const fixture = productionScaleCoordinationFixture("goal-fixture"); const before = structuredClone(fixture);