diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 7ad2211799..7407792c25 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -6,6 +6,11 @@ Agent Session calendar reports are implemented. Installation, an enabled subscription and a natural scheduled delivery require their own readback evidence. Preparing an identity enables neither monitoring nor reports. +The same local calendar can explicitly repeat an already completed native Task +in its original managed Session. This backend/configuration slice is described +below; arbitrary new tasks from natural language and the unified Agent UI +remain separate, unfinished acceptance. + The steward becomes a persistent LoopX-managed Agent. Its original private conversation remains the user's entry point. Native monitor Todos, Turn admission, scheduling and periodic-report publication own recurring work. @@ -732,3 +737,102 @@ alone does not activate a subscription or deliver a report. The opt-in original steward report path binds it to the admitted calendar window, native generation and private full-body delivery; installed adoption, current configuration and the original production route still need independent verification. + + +## Repeat an existing native Task in a managed Session + +Use this optional mode for “repeat this completed task on a calendar.” It uses +an existing Goal, registered Agent and exact local managed Session. It creates +no Goal, executor, recurring cron or separate task database. The source Task +must already be completed, belong to that Agent and retain an independent argv +validation declaration. Open, deferred, blocked, foreign and unvalidated tasks +are rejected; configuring a calendar does not resume them or install commands. +Report-generation and calendar-generated Tasks are already window-bound and +cannot be sources themselves. Use the existing report calendar for reports. +This uses the Goal's existing single calendar subscription. Selecting this +mode replaces that Goal's previous subscription after the pending-window +guard; it does not add another schedule beside an enabled report calendar. +The original validator must check the repeated task's current expected +outcome; a check pinned to an old output does not establish fresh business +work. Scheduling preserves that declaration rather than improving its proof. + +The existing Goal capability configuration transaction accepts: + +```json +{ + "goal_id": "example-project", + "capability_id": "periodic_report", + "configuration": { + "enabled": true, + "profile_preset": "session-task", + "route_ref": "managed-session:11111111111111111111111111111111", + "timezone": "UTC", + "schedule": { + "schema_version": "periodic_report_schedule_v0", + "schedule_id": "repeat-completed-task", + "timezone": "UTC", + "rrule": "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", + "start_date": "2030-01-01" + }, + "task": {"source_todo_id": "todo_completed_example"} + } +} +``` + +Replace the example with the actual existing identities and requested time. +Prepare through `/api/chat/goal-configuration/preview`, then apply the same +body and its `expected_plan_revision` through +`/api/chat/goal-configuration/apply`. The existing Goal settings JSON editor +uses this transaction; read back the saved configuration and the managed +Session's `scheduled_task` inspection. A stale preview or changed identity is +rejected. This is an explicit Goal override: a machine default cannot enroll +another conversation. `session-task` is a calendar execution mode, not a report +renderer preset; it does not invoke the progress-report source or editorial +format. Task descriptions, scripts and validation argv cannot be supplied in +this configuration. + +At the first due boundary, the original idle Chat pump freezes the source +Task's declaration into the existing cadence attempt. It creates one finite +native Task for that window, inheriting the source Task's scope, repository, +capability requirements, decision requirements and independent validation. +TypeScript admission rechecks those declarations, the exact Session, Goal +lifetime and current configuration. The same native Turn worker executes and +settles it. Successful Task validation, native commit and the complete original +Core answer must all read back before that period is acknowledged. This Task +completion receipt is distinct from report/provider publication; it never +claims an external message was sent. Source and previous-period Tasks may be +read from canonical completed history after archival. + +Before the first deadline, while the Session has accepted user work, or while +the schedule is disabled, this path starts no task model. Ordinary calendar +and source reads still have local IO costs. Each actual execution uses native +admission, a Task lease, independent validation and quota settlement. Token +usage and fees depend on the configured host; the calendar makes no savings +claim. Duplicate observations and a lost completion acknowledgement reconcile +the existing Task and Core message without another model execution or answer. + +Set `enabled=false` to pause new work and delivery; retain the rest of the +configuration to resume that plan. Clearing the override after a known failed +independent task-postcondition validation retains that Task and retires only +that window, after verified +configuration apply. Re-enabling cannot replay the failed period; the next +independent due boundary can execute. Unknown, partial or interrupted work is +retained for the existing native recovery owner; there is no automatic model +replay. A template/source change during an unfinished period cannot silently +replace the frozen task. Clearing or disabling this mode preserves ordinary +Chat, attached Sessions, existing reports and their histories. + +This slice does not interpret conversation text as a subscription, configure +an arbitrary future task that has no completed native source, send to an +external destination, or finish the unified managed/attached Agent journey. +Those remain part of the existing recurring-task work. Tests use real HTTP, +File/SQLite, native Tasks/Turns and an independently checked synthetic model +artifact; they do not establish live-model comprehension, installed adoption, +a natural production deadline or a full frontend acceptance. + +For rollback, pause or cancel through a runtime that understands this mode. +Older runtimes do not understand its Task completion receipt; downgrading that +Goal's calendar owner with retained Task windows is unqualified. Keep the +compatible reader and original evidence, rather than deleting the journal to +make an older reader accept it. Ordinary feature-off Goals create no new +calendar Task or Task receipt and retain their existing rollback path. diff --git a/loopx/capabilities/native_chat/session_calendar_tasks.py b/loopx/capabilities/native_chat/session_calendar_tasks.py new file mode 100644 index 0000000000..733d395a70 --- /dev/null +++ b/loopx/capabilities/native_chat/session_calendar_tasks.py @@ -0,0 +1,279 @@ +"""Calendar IO for ordinary native Tasks in an existing managed Session. + +The existing Goal configuration, cadence journal, Todo, Turn, validator and +Core conversation are the owners. This adapter adds no worker or task store. +An explicit task declaration is required; conversation prose is never parsed +here as consent, a schedule or an executable command. +""" +from __future__ import annotations + +from datetime import datetime, timezone +import json +import re + +from ...agent_registry import load_goal_from_registry +from ...control_plane.digest_envelope import sha256_envelope +from ...control_plane.todos.contract import compact_todo_text, decision_scope_metadata_value +from ...control_plane.todos.completion_validation_store import read_completion_validation_declaration +from ...file_lock import exclusive_file_lock +from ...registry import atomic_write_json +from ..periodic_report.cadence import report_cadence_next_due_at +from ..periodic_report.cadence_journal import ( + admit_cadence_window, cadence_journal_path, read_cadence_journal, + record_cadence_task_completion, +) +from ..periodic_report.goal_configuration import configuration_summary +from ..periodic_report.machine_defaults import resolve_active_goal_periodic_report_subscription +from ..periodic_report.pending_intent import _canonical_digest +from ..periodic_report.private_generation import attempt_directory +from .steward_commands import native_command +from .steward_reports import ManagedSessionReports + + +class SessionCalendarTasks(ManagedSessionReports): + def subscription(self, session_id, *, expected_revision=None): + actor = self.selected(session_id)["actor"] + goal = load_goal_from_registry(self.controller.registry_path, actor["goal_id"]) + config = configuration_summary(goal) + if (not config or config.get("profile_preset") != "session-task" + or config.get("route_ref") != "managed-session:" + session_id): + return None + subscription = resolve_active_goal_periodic_report_subscription(goal) + if subscription and expected_revision is not None and subscription["effective_revision"] != expected_revision: + return None + return subscription + + def _definition(self, window): + path = attempt_directory(self.root, window) / "task.json" + if not path.is_file(): + return None + value = json.loads(path.read_text(encoding="utf-8")) + if (not isinstance(value, dict) or value.get("window") != window or value.get("digest") != _canonical_digest( + {k: v for k, v in value.items() if k != "digest"})): + raise ValueError("the frozen calendar task changed") + actor = value.get("actor") + if not isinstance(actor, dict) or self.selected(actor.get("session_id"))["actor"] != actor: + raise ValueError("the original calendar task actor changed") + return value + + def _frozen_actor(self, window): + value = self._definition(window) + return value["actor"] if value else None + + def _todo(self, actor, window): + from .steward_monitor import StewardMonitor + from ...control_plane.todos.list_readback import list_goal_todos + rows = [row for row in StewardMonitor(self.controller).todos(actor) + if row.get("target_key") == window["window_id"]] + if not rows: + # Active-list absence never permits another execution. A finished + # task may already be in the canonical completed-history view. + history = list_goal_todos(registry_path=self.controller.registry_path, goal_id=actor["goal_id"], + runtime_root_arg=str(self.root), role="agent", status="done", read_scope="completed_history", limit=None) + rows = [row for row in history["todos"] if row.get("target_key") == window["window_id"]] + if len(rows) > 1: + raise ValueError("the calendar window has conflicting original Tasks") + return rows[0] if rows else None + + def source_task(self, actor, task): + """Read a completed native Task; configuration supplies no host argv.""" + from ...control_plane.todos.list_readback import list_goal_todos + row = list_goal_todos(registry_path=self.controller.registry_path, goal_id=actor["goal_id"], + runtime_root_arg=str(self.root), todo_id=task["source_todo_id"], role="agent").get("todo") + if row is None: + row = list_goal_todos(registry_path=self.controller.registry_path, goal_id=actor["goal_id"], + runtime_root_arg=str(self.root), todo_id=task["source_todo_id"], role="agent", + status="done", read_scope="completed_history").get("todo") + if (not row or row.get("status") != "done" or row.get("claimed_by") != actor["agent_id"] + or row.get("task_class") != "advancement_task" or row.get("completion_validation_required") is not True + or row.get("action_kind") == "generate_periodic_report" + or re.fullmatch(r"cadence_[a-f0-9]{64}", row.get("target_key") or "") + or actor["agent_id"] in row.get("excluded_agents", [])): + raise ValueError("select this Agent's existing completed Task with independent argv validation") + declaration = read_completion_validation_declaration(runtime_root=self.root, + goal_id=actor["goal_id"], todo_id=row["todo_id"], expected_digest=row["completion_validation_sha256"]) + if not declaration or not declaration.get("validation_command_argv"): + raise ValueError("the existing source task has no independent argv validation") + return row, declaration + + def admission(self, request): + actor = request["actor"] + subscription, window = self.current_window(actor["session_id"]) + if (not subscription or not window or window["window_id"] != request["schedule_window_id"] + or subscription["effective_revision"] != window["subscription_revision"]): + raise ValueError("the original calendar task subscription changed") + definition = self._definition(window) + todo = self._todo(actor, window) + if not definition or not todo: + raise ValueError("the original finite calendar task is unavailable") + source_task, validation = self.source_task(actor, subscription["task"]) + declaration = read_completion_validation_declaration(runtime_root=self.root, + goal_id=actor["goal_id"], todo_id=todo["todo_id"], + expected_digest=todo.get("completion_validation_sha256")) + if not declaration or declaration != validation or declaration != definition["validation"]: + raise ValueError("the original calendar task's independent validation changed") + return {"subscription": subscription, "window": window, "definition": definition, "todo": todo, "source_task": source_task} + + def _completion(self, window): + """Task DONE alone is insufficient: read the original native commit/body.""" + definition = self._definition(window) + if not definition: + return None + actor = definition["actor"] + todo = self._todo(actor, window) + completed_key = (todo or {}).get("completion_turn_key") + client = ("managed." + completed_key.removeprefix("managed-session.") + if isinstance(completed_key, str) and completed_key.startswith("managed-session.") + else "managed.task." + window["window_id"]) + turn = self.controller.store.turn_for_client(actor["session_id"], client) + request = (turn or {}).get("loopx_request") or {} + response = (turn or {}).get("response") or {} + if (not todo or not turn or turn["status"] != "completed" or todo["status"] != "done" + or request.get("work_kind") != "scheduled_task" or request.get("actor") != actor + or request.get("todo_id") != todo["todo_id"] or request.get("schedule_window_id") != window["window_id"] + or request.get("schedule_subscription_revision") != window["subscription_revision"] + or todo.get("completion_turn_key") != request.get("turn_instance_id") + or (response.get("native_loopx") or {}).get("status") != "committed"): + return None + if not any(row["kind"] == "native_loopx.result" and row["payload"].get("status") == "committed" + for row in self.controller.store.events_after(actor["session_id"], turn["turn_id"], None)): + return None + body = response.get("message") + rows = [row for row in self.controller.store.messages(actor["session_id"]) + if row["message_id"] == "managed." + turn["turn_id"] + ".completed"] + if (not isinstance(body, str) or not body.strip() or len(rows) != 1 + or rows[0]["role"] != "agent" or rows[0]["turn_id"] != turn["turn_id"] or rows[0]["text"] != body): + return None + return {"todo_id": todo["todo_id"], "turn_instance_id": request["turn_instance_id"], + "message_id": rows[0]["message_id"], "message_sha256": sha256_envelope(body.encode("utf-8")), + "completed_at": rows[0]["created_at"]} + + def _published_window(self, window): + # Compatibility with the shared configuration fence; this remains a + # Task completion, never an invented report/provider publication. + return self._completion(window) is not None + + def _known_generation_failure(self, actor, window): + turn = self.controller.store.turn_for_client(actor["session_id"], "managed.task." + window["window_id"]) + request = (turn or {}).get("loopx_request") or {} + native = ((turn or {}).get("response") or {}).get("native_loopx") or {} + todo = self._todo(actor, window) + validation, receipt, effects = native.get("validation") or {}, native.get("receipt") or {}, native.get("effects") or {} + if (turn and turn["status"] == "completed" and native.get("status") == "failed" + and todo and todo.get("status") in {"open", "blocked"} + and native.get("result_kind") == "validation_failed" and native.get("validation_stage") == "task_postcondition" + and validation.get("schema_version") == "loopx_turn_task_validation_v0" + and validation.get("ok") is False and validation.get("validator_kind") == "command" + and receipt.get("ok") is True and receipt.get("failed_phase") == "validation" + and receipt.get("completed_phases") == ["host_execute", "typed_result"] + and receipt.get("lineage") == {"goal_id": actor["goal_id"], "agent_id": actor["agent_id"], "todo_id": todo["todo_id"]} + and effects.get("state_written") is False and effects.get("quota_spent") is False + and request.get("work_kind") == "scheduled_task" and request.get("actor") == actor + and request.get("todo_id") == todo["todo_id"] + and request.get("schedule_window_id") == window["window_id"] + and request.get("schedule_subscription_revision") == window["subscription_revision"]): + return {"turn_id": turn["turn_id"], "todo_id": request["todo_id"]} + return None + + def _cancel_failed_window(self, actor, window, failure, *, cadence_locked=False): + from ..periodic_report.cadence_journal import ( + _retire_failed_cadence_window_locked, retire_failed_cadence_window, + ) + if self._known_generation_failure(actor, window) != failure: + raise ValueError("the original failed calendar task changed before cancellation") + native_command(self.controller, ["todo", "update", "--goal-id", actor["goal_id"], + "--todo-id", failure["todo_id"], "--agent-id", actor["agent_id"], + "--update-operation-id", "cancel.task." + window["window_id"], "--status", "blocked", + "--clear-resume-when", "--reason", + "Owner cancelled this known failed calendar task; retain original work, effects and failure evidence, not completion.", "--execute"]) + retire = _retire_failed_cadence_window_locked if cadence_locked else retire_failed_cadence_window + retire(runtime_root=self.root, goal_id=actor["goal_id"], expected_window_id=window["window_id"], + source_receipt_id=failure["turn_id"], reason="owner_cancelled_failed_task") + + def frontier(self, session_id): + from .steward_managed_mode import ManagedSessionTurn + actor = self.selected(session_id)["actor"] + subscription = self.subscription(session_id) + if not subscription or not subscription.get("schedule"): + return None, False + journal = read_cadence_journal(runtime_root=self.root, goal_id=actor["goal_id"]) + if (journal and not journal.get("task_completion") + and journal["window"]["profile_ref"]["profile_id"] == "session_task"): + receipt = self._completion(journal["window"]) + if receipt: + record_cadence_task_completion(runtime_root=self.root, goal_id=actor["goal_id"], + expected_window_id=journal["window"]["window_id"], receipt=receipt) + admitted = admit_cadence_window(runtime_root=self.root, goal_id=actor["goal_id"], + agent_id=actor["agent_id"], subscription=subscription, now=datetime.now(timezone.utc), + subscription_resolver=lambda: self.subscription(session_id)) + if admitted["status"] != "pending": + return None, False + window = admitted["window"] + with exclusive_file_lock(cadence_journal_path(self.root, actor["goal_id"]), operation="prepare_calendar_task"): + current = self.subscription(session_id, expected_revision=window["subscription_revision"]) + live_journal = read_cadence_journal(runtime_root=self.root, goal_id=actor["goal_id"]) + if current is None or not live_journal or live_journal["window"] != window: + return None, False + source_task, validation = self.source_task(actor, current["task"]) + path = attempt_directory(self.root, window) / "task.json" + definition = self._definition(window) + if definition is None: + text = compact_todo_text(f"[{source_task.get('priority') or 'P1'}] Scheduled task for {window['start_at']} to {window['due_at']}: {source_task.get('title') or source_task['text']}") + definition = {"actor": actor, "window": window, "task": current["task"], "todo_text": text, + "source_task": source_task, "validation": validation} + definition["digest"] = _canonical_digest(definition) + atomic_write_json(path, definition) + todo = self._todo(actor, window) + if todo is None: + argv = ["todo", "add", "--goal-id", actor["goal_id"], "--role", "agent", + "--operation-id", window["window_id"], "--text", definition["todo_text"], + "--claimed-by", actor["agent_id"], + "--priority", source_task.get("priority") or "P1", "--task-class", "advancement_task", + "--target-key", window["window_id"], "--continuation-policy", "same_agent_non_delivery", + "--validation-command-json", json.dumps(definition["validation"]["validation_command_argv"]), "--execute"] + for key, flag in {"action_kind": "--action-kind", "task_domain": "--task-domain", "task_repository": "--task-repository", + "capability_binding_ref": "--capability-binding-ref"}.items(): + if source_task.get(key): + argv += [flag, source_task[key]] + for key, flag in {"required_write_scopes": "--required-write-scope", "required_capabilities": "--required-capability", + "target_capabilities": "--target-capability", + "excluded_agents": "--excluded-agent"}.items(): + for item in source_task.get(key, []): + argv += [flag, item] + for scope in source_task.get("required_decision_scopes", []): + argv += ["--required-decision-scope", decision_scope_metadata_value(scope)] + for key, flag in {"validation_label": "--validation-label", "validation_timeout_seconds": "--validation-timeout-seconds"}.items(): + if validation.get(key) is not None: + argv += [flag, str(validation[key])] + native_command(self.controller, argv) + todo = self._todo(actor, window) + if not todo or todo["status"] != "open": + return None, False + return ManagedSessionTurn(self.controller).submit(session_id, agent_id=actor["agent_id"], + operation_id="task." + window["window_id"], todo_id=todo["todo_id"], scheduled_window=window) + + def health(self, session_id): + configured = False + try: + actor = self.selected(session_id)["actor"] + goal = load_goal_from_registry(self.controller.registry_path, actor["goal_id"]) + config = configuration_summary(goal) + if not config or config.get("profile_preset") != "session-task" or config.get("route_ref") != "managed-session:" + session_id: + return {"available": True, "configured": False, "enabled": False} + configured = True + journal = read_cadence_journal(runtime_root=self.root, goal_id=actor["goal_id"]) + receipt = self._completion(journal["window"]) if journal else None + client = ("managed." + receipt["turn_instance_id"].removeprefix("managed-session.") + if receipt else "managed.task." + journal["window"]["window_id"] if journal else None) + turn = self.controller.store.turn_for_client(session_id, client) if client else None + native = ((turn or {}).get("response") or {}).get("native_loopx") or {} + return {"available": True, "configured": True, "enabled": bool(self.subscription(session_id)), + "schedule": config.get("schedule"), "task": config.get("task"), + "next_due_at": report_cadence_next_due_at(config.get("schedule"), now=datetime.now(timezone.utc)) if config.get("enabled") else None, + "retained_window_due_at": journal["window"]["due_at"] if journal else None, + "completion_verified": bool(receipt), "completion": receipt, + "task_id": (turn or {}).get("loopx_request", {}).get("todo_id"), + "turn_id": (turn or {}).get("turn_id"), "native_result_status": native.get("status"), + "result_kind": native.get("result_kind")} + except (OSError, ValueError, KeyError, TypeError): + return {"available": False, "configured": configured, "error_code": "managed_calendar_task_unavailable"} diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index 83f502b86d..99c46f95ff 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -273,6 +273,9 @@ def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, report = {"report_subscription": subscription, "report_window": window, "report_todo": next((row for row in StewardMonitor(self.controller).todos(actor) if row["todo_id"] == request.get("todo_id")), None)} + if request.get("work_kind") == "scheduled_task": + from .session_calendar_tasks import SessionCalendarTasks + report["scheduled_task"] = SessionCalendarTasks(self.controller).admission(request) return ChatConversationBindings._core("collaboration.session.managed_turn.admit", { "session": session, "request": request, "turn": turn, "goal": StewardManagedMode(self.controller, None)._goal(actor["goal_id"]), @@ -302,16 +305,26 @@ def inspect(self, session_id: str, *, agent_id: str | None = None, result["reason"] = str(exc) if session.get("managed_agent_id"): from .steward_reports import ManagedSessionReports + from .session_calendar_tasks import SessionCalendarTasks result["report"] = ManagedSessionReports(self.controller).health(session_id) + task = SessionCalendarTasks(self.controller).health(session_id) + if task.get("configured"): + result["scheduled_task"] = task return result def submit(self, session_id: str, *, agent_id: str, operation_id: str, todo_id: str | None = None, monitor_deadline: str | None = None, - report_window: dict[str, Any] | None = None) -> tuple[dict[str, Any], bool]: + report_window: dict[str, Any] | None = None, + scheduled_window: dict[str, Any] | None = None) -> tuple[dict[str, Any], bool]: from ...attached_session import managed_session_goal_commit controller = self.controller with controller._session_adapter_lock(session_id): request = self._request(session_id, agent_id, operation_id, todo_id) + if scheduled_window is not None: + if report_window is not None or monitor_deadline is not None or not todo_id: + raise ValueError("a calendar task requires its original finite Todo") + request.update(work_kind="scheduled_task", schedule_window_id=scheduled_window["window_id"], + schedule_subscription_revision=scheduled_window["subscription_revision"]) if report_window is not None: if monitor_deadline is not None or not todo_id: raise ValueError("a report requires its original finite generation Todo") @@ -341,6 +354,7 @@ def submit(self, session_id: str, *, agent_id: str, operation_id: str, accepted = self.store.accept_managed_turn(session_id, client_turn_id="managed." + operation_id, message="Advance the currently admitted native LoopX work.", display_message="生成本期定时报告。" if report_window is not None else + "执行本期定时任务。" if scheduled_window is not None else "检查当前 Goal 原生任务状态。" if monitor_deadline is not None else "推进一次 LoopX 任务。", origin="native_loopx", loopx_execution=True, loopx_request=request) try: @@ -377,6 +391,8 @@ def prepare_validation(self, request: dict[str, Any], native_request: dict[str, if not selected.get("todo_id"): raise ManagedTaskValidationUnavailable("select an existing native task with independent validation") self.validation_basis = self._validation_source(request, selected["todo_id"]) + if re.fullmatch(r"cadence_[a-f0-9]{64}", self.validation_basis.get("target_key") or "") and request.get("work_kind") != "scheduled_task": + raise ValueError("the original calendar Task must execute through its scheduled Session admission") declaration = read_completion_validation_declaration(runtime_root=self.controller.coordination_runtime_root, goal_id=request["actor"]["goal_id"], todo_id=selected["todo_id"], expected_digest=self.validation_basis["completion_validation_sha256"]) diff --git a/loopx/capabilities/native_chat/steward_monitor.py b/loopx/capabilities/native_chat/steward_monitor.py index cdcf615f53..2ed450c194 100644 --- a/loopx/capabilities/native_chat/steward_monitor.py +++ b/loopx/capabilities/native_chat/steward_monitor.py @@ -349,6 +349,16 @@ def pump_managed_session_monitors(controller: Any, *, cancelled=lambda: False) - service.admit(request) goal = load_goal_from_registry(controller.registry_path, session["goal_id"]) report_config = (goal or {}).get("control_plane", {}).get("periodic_report", {}) + if (report_config.get("enabled") is True and report_config.get("profile_preset") == "session-task" + and report_config.get("route_ref") == "managed-session:" + session["session_id"]): + from .session_calendar_tasks import SessionCalendarTasks + try: + accepted, created = SessionCalendarTasks(controller).frontier(session["session_id"]) + if created: + changed.append(accepted) + continue + except (OSError, ValueError, KeyError, TypeError, RuntimeError): + _LOG.warning("A managed calendar task is unavailable; its original window and work are retained") if (report_config.get("enabled") is True and report_config.get("profile_preset") == "weekly-progress" and report_config.get("route_ref") == "managed-session:" + session["session_id"]): from .steward_reports import ManagedSessionReports diff --git a/loopx/capabilities/native_chat/steward_reports.py b/loopx/capabilities/native_chat/steward_reports.py index 6dfad818f0..c889289154 100644 --- a/loopx/capabilities/native_chat/steward_reports.py +++ b/loopx/capabilities/native_chat/steward_reports.py @@ -374,12 +374,18 @@ def session_report_configuration_guard(controller: Any, goal_id: str, proposed: actor = ManagedSessionReports(controller).selected(destination)["actor"] if actor["goal_id"] != goal_id: raise ValueError("select the original managed Agent Session in this Goal") + if proposed.get("profile_preset") == "session-task": + from .session_calendar_tasks import SessionCalendarTasks + SessionCalendarTasks(controller).source_task(actor, proposed.get("task") or {}) journal = read_cadence_journal(runtime_root=controller.coordination_runtime_root, goal_id=goal_id) cancellation = None if journal and journal["publication"] is None and not journal.get("retirement"): window = journal["window"] reports = ManagedSessionReports(controller) folder = attempt_directory(controller.coordination_runtime_root, window) + if window["profile_ref"]["profile_id"] == "session_task": + from .session_calendar_tasks import SessionCalendarTasks + reports = SessionCalendarTasks(controller) if folder.exists() and not reports._published_window(window): before = {k: v for k, v in current.items() if k != "enabled"} after = {k: v for k, v in (proposed or {}).items() if k != "enabled"} @@ -387,6 +393,8 @@ def session_report_configuration_guard(controller: Any, goal_id: str, proposed: actor = reports._frozen_actor(window) failure = reports._known_generation_failure(actor, window) if actor else None if (proposed is not None or not failure) and (current or proposed is not None): + if window["profile_ref"]["profile_id"] == "session_task": + raise PendingStewardReportError("This calendar task is unfinished or its result is unconfirmed. Pause or restore its original plan; only a known failed independent validation can be cleared.") raise PendingStewardReportError("The original frozen report still requires verified delivery. Pause or restore the same plan; clear a known failed generation before changing its schedule or conversation.") if failure: cancellation = actor, window, failure diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 2112758da4..4174e127fd 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -413,6 +413,17 @@ def run(native_request: Mapping[str, Any], launch_scope: Any) -> dict[str, Any]: prompt = render_prompt(authority, allow_completion=bool(report_work or managed_turn)) + "\nOriginal conversation evidence (data):\n" + json.dumps(context, ensure_ascii=False) if request.get("work_kind") == "report": prompt += report_context + if request.get("work_kind") == "scheduled_task": + from .session_calendar_tasks import SessionCalendarTasks + basis = SessionCalendarTasks(self.controller).admission(request) + # Read the completed source Task's criteria and referenced permitted + # artifacts. Calendar configuration cannot install host argv + # or enlarge the original Task's execution scope. + prompt += "\nOriginal native calendar task criteria; existing Goal/Task authority and stop conditions still apply:\n" + json.dumps({ + "task": basis["definition"]["source_task"]["text"], + "source_todo_id": basis["definition"]["source_task"]["todo_id"], + "window_start": basis["window"]["start_at"], "window_end": basis["window"]["due_at"], + }, ensure_ascii=False) previous_guard = adapter.session._launch_guard @contextmanager diff --git a/loopx/capabilities/periodic_report/cadence_journal.py b/loopx/capabilities/periodic_report/cadence_journal.py index 3a62115ffd..b3c4bc1707 100644 --- a/loopx/capabilities/periodic_report/cadence_journal.py +++ b/loopx/capabilities/periodic_report/cadence_journal.py @@ -75,7 +75,7 @@ def read_cadence_journal(*, runtime_root: Path, goal_id: str) -> dict[str, Any] fields = { "schema_version", "goal_id", "window", "admitted_at", "publication", } - if not isinstance(value, dict) or set(value) - {"supersedes", "retirement"} != fields or value["schema_version"] != JOURNAL_SCHEMA or value["goal_id"] != goal_id: + if not isinstance(value, dict) or set(value) - {"supersedes", "retirement", "task_completion"} != fields or value["schema_version"] != JOURNAL_SCHEMA or value["goal_id"] != goal_id: raise ValueError("cadence journal contract is invalid") if "supersedes" in value and not re.fullmatch(r"cadence_[0-9a-f]{64}", str(value["supersedes"])): raise ValueError("cadence journal predecessor is invalid") @@ -90,16 +90,54 @@ def read_cadence_journal(*, runtime_root: Path, goal_id: str) -> dict[str, Any] ): raise ValueError("cadence journal publication reference is invalid") retirement = value.get("retirement") - if retirement is not None and (publication is not None or not isinstance(retirement, dict) + completion = value.get("task_completion") + if completion is not None and (publication is not None or not isinstance(completion, dict) + or window["profile_ref"].get("profile_id") != "session_task" + or set(completion) != {"todo_id", "turn_instance_id", "message_id", "message_sha256", "completed_at"} + or any(not isinstance(item, str) or not item for item in completion.values()) + or not ENVELOPED_SHA256_PATTERN.fullmatch(completion["message_sha256"])): + raise ValueError("cadence task completion reference is invalid") + if retirement is not None and (publication is not None or completion is not None or not isinstance(retirement, dict) or set(retirement) != {"reason", "source_receipt_id", "retired_at"} - or retirement["reason"] != "owner_cancelled_failed_generation" + or retirement["reason"] not in {"owner_cancelled_failed_generation", "owner_cancelled_failed_task"} or any(not isinstance(item, str) or not item for item in retirement.values())): raise ValueError("cadence journal retirement reference is invalid") return value +def record_cadence_task_completion(*, runtime_root: Path, goal_id: str, + expected_window_id: str, receipt: dict[str, str]) -> dict[str, Any]: + """Checkpoint the native owner's independently read Task and Core reply. + + This is a finite task completion, not a report publication. The caller + verifies native commit, Task validation and the exact original Core body. + Reconciliation after a lost write must supply the same receipt. + """ + path = cadence_journal_path(runtime_root, goal_id) + with exclusive_file_lock(path, policy=LockAcquisitionPolicy.MUTATION, + operation="complete_managed_calendar_task"): + current = read_cadence_journal(runtime_root=runtime_root, goal_id=goal_id) + if (not current or current["window"]["window_id"] != expected_window_id + or current["window"]["profile_ref"].get("profile_id") != "session_task" + or current["publication"] is not None or current.get("retirement")): + raise ValueError("the original unpublished calendar task changed") + if current.get("task_completion"): + if current["task_completion"] != receipt: + raise ValueError("the calendar task completion changed its original result") + return current + proposed = {**current, "task_completion": receipt} + # Validate before writing using the same receipt shape as readback. + if (set(receipt) != {"todo_id", "turn_instance_id", "message_id", "message_sha256", "completed_at"} + or any(not isinstance(item, str) or not item for item in receipt.values()) + or not ENVELOPED_SHA256_PATTERN.fullmatch(receipt["message_sha256"])): + raise ValueError("cadence task completion reference is invalid") + atomic_write_json(path, proposed) + return proposed + + def retire_failed_cadence_window(*, runtime_root: Path, goal_id: str, - expected_window_id: str, source_receipt_id: str) -> dict[str, Any]: + expected_window_id: str, source_receipt_id: str, + reason: str = "owner_cancelled_failed_generation") -> dict[str, Any]: """Retain an explicitly cancelled known failure; never invent publication. The original execution owner must prove a terminal failed generation with @@ -110,22 +148,24 @@ def retire_failed_cadence_window(*, runtime_root: Path, goal_id: str, with exclusive_file_lock(path, policy=LockAcquisitionPolicy.MUTATION, operation="retire_failed_report_window"): return _retire_failed_cadence_window_locked(runtime_root=runtime_root, goal_id=goal_id, - expected_window_id=expected_window_id, source_receipt_id=source_receipt_id) + expected_window_id=expected_window_id, source_receipt_id=source_receipt_id, reason=reason) def _retire_failed_cadence_window_locked(*, runtime_root: Path, goal_id: str, - expected_window_id: str, source_receipt_id: str) -> dict[str, Any]: + expected_window_id: str, source_receipt_id: str, + reason: str = "owner_cancelled_failed_generation") -> dict[str, Any]: """Reuse the transition while the settings owner holds the cadence lock.""" - if not isinstance(source_receipt_id, str) or not source_receipt_id: + if (not isinstance(source_receipt_id, str) or not source_receipt_id + or reason not in {"owner_cancelled_failed_generation", "owner_cancelled_failed_task"}): raise ValueError("failed cadence retirement requires its original receipt") current = read_cadence_journal(runtime_root=runtime_root, goal_id=goal_id) - if not current or current["window"]["window_id"] != expected_window_id or current["publication"] is not None: + if not current or current["window"]["window_id"] != expected_window_id or current["publication"] is not None or current.get("task_completion"): raise ValueError("the unpublished failed report window changed") if current.get("retirement"): - if current["retirement"]["source_receipt_id"] != source_receipt_id: + if current["retirement"]["source_receipt_id"] != source_receipt_id or current["retirement"]["reason"] != reason: raise ValueError("the failed window retirement changed its source") return current - current["retirement"] = {"reason": "owner_cancelled_failed_generation", + current["retirement"] = {"reason": reason, "source_receipt_id": source_receipt_id, "retired_at": datetime.now(timezone.utc).isoformat()} atomic_write_json(cadence_journal_path(runtime_root, goal_id), current) return current @@ -182,20 +222,32 @@ def _admit_cadence_window_locked( if subscription.get("enabled") is not True or subscription.get("schedule") is None: return {"status": "disabled", "mutated": False, "window": None} calculated = report_cadence_window(subscription["schedule"], now=now) - from .presets import build_periodic_report_preset_activation - activation = build_periodic_report_preset_activation(str(subscription.get("profile_preset") or "")) - profile = activation.get("profile") - if activation.get("active") is not True or not isinstance(profile, Mapping) or ( - "cadence_due" not in profile.get("trigger_policy", {}).get("enabled_kinds", []) - ): - raise ValueError("periodic-report profile does not support calendar reports") + if subscription.get("profile_preset") == "session-task": + from .goal_configuration import managed_session_report_destination, normalize_session_task + if subscription.get("source") != "goal_override" or not managed_session_report_destination(subscription): + raise ValueError("calendar tasks require an explicit original managed Session") + normalize_session_task(subscription.get("task")) + # Reuse the native calendar, without advertising a report renderer or + # a report source adapter that an ordinary Task never invokes. + profile_ref = {"profile_id": "session_task", "profile_version": "v1"} + profile_ref["profile_digest"] = _digest(profile_ref) + trigger_policy = {"enabled_kinds": ["cadence_due"], "minimum_interval_seconds": 0} + else: + from .presets import build_periodic_report_preset_activation + activation = build_periodic_report_preset_activation(str(subscription.get("profile_preset") or "")) + profile = activation.get("profile") + if activation.get("active") is not True or not isinstance(profile, Mapping) or ( + "cadence_due" not in profile.get("trigger_policy", {}).get("enabled_kinds", []) + ): + raise ValueError("periodic-report profile does not support calendar reports") + profile_ref = {"profile_id": profile["profile_id"], "profile_version": profile["profile_version"], + "profile_digest": activation["profile_digest"]} + trigger_policy = profile["trigger_policy"] proposed = { "schema_version": CADENCE_WINDOW_SCHEMA, "goal_id": goal_id, "agent_id": agent_id, "subscription_revision": subscription.get("effective_revision"), - "profile_ref": {"profile_id": profile["profile_id"], - "profile_version": profile["profile_version"], - "profile_digest": activation["profile_digest"]}, - "trigger_policy": profile["trigger_policy"], + "profile_ref": profile_ref, + "trigger_policy": trigger_policy, **calculated, } if calculated is not None else None if proposed is not None: @@ -206,7 +258,7 @@ def _admit_cadence_window_locked( supersedes = None if previous is not None: old = previous["window"] - if previous["publication"] is None and not previous.get("retirement"): + if previous["publication"] is None and not previous.get("retirement") and not previous.get("task_completion"): # Import lazily to keep the pure calendar evaluator independent # of subscription and generation composition. from .incremental import read_periodic_report_publication_cursor @@ -237,6 +289,11 @@ def _admit_cadence_window_locked( if proposed is None or old["due_at"] >= proposed["due_at"]: return {"status": "retired", "mutated": False, "window": old} supersedes = _archive_predecessor(path, previous) + if previous.get("task_completion"): + if proposed is not None and old["due_at"] >= proposed["due_at"]: + return {"status": "already_completed", "mutated": False, "window": old} + if proposed is not None: + supersedes = _archive_predecessor(path, previous) # A route/profile edit is not a second report for the same due # boundary. Clock rollback likewise cannot replay a delivered one. if proposed is not None and supersedes is None and old["due_at"] >= proposed["due_at"]: diff --git a/loopx/capabilities/periodic_report/goal_configuration.py b/loopx/capabilities/periodic_report/goal_configuration.py index 7bbf2e989c..a6ef4e0929 100644 --- a/loopx/capabilities/periodic_report/goal_configuration.py +++ b/loopx/capabilities/periodic_report/goal_configuration.py @@ -19,13 +19,29 @@ def managed_session_report_destination(configuration: Mapping[str, Any]) -> str Goal lifetime, subscription, Task and original worker on every dispatch. """ route = configuration.get("route_ref") - if configuration.get("profile_preset") == "weekly-progress" and isinstance(route, str): + if configuration.get("profile_preset") in {"weekly-progress", "session-task"} and isinstance(route, str): match = re.fullmatch(r"managed-session:([a-f0-9]{32})", route) if match: return match[1] return None +def normalize_session_task(value: object) -> dict[str, Any]: + """Codec for an existing native Task reference, not an NL classifier. + + The existing Todo validator and managed Session admission still own + execution. A description cannot grant a workspace, recipient or command. + """ + from ...control_plane.todos.contract import normalize_todo_id + + if not isinstance(value, Mapping) or set(value) != {"source_todo_id"}: + raise ValueError("session task requires one existing source_todo_id; configuration cannot install commands") + todo_id = normalize_todo_id(value["source_todo_id"]) + if not todo_id: + raise ValueError("session task requires an exact completed native source Todo") + return {"source_todo_id": todo_id} + + def configuration_summary(goal: Mapping[str, Any]) -> dict[str, Any] | None: control_plane = goal.get("control_plane") periodic = ( @@ -37,7 +53,7 @@ def configuration_summary(goal: Mapping[str, Any]) -> dict[str, Any] | None: def normalize_configuration(value: Mapping[str, Any]) -> dict[str, Any]: - allowed = {"enabled", "profile_preset", "route_ref", "timezone", "schedule"} + allowed = {"enabled", "profile_preset", "route_ref", "timezone", "schedule", "task"} unknown = sorted(set(value) - allowed) if unknown: raise ValueError( @@ -59,14 +75,22 @@ def normalize_configuration(value: Mapping[str, Any]) -> dict[str, Any]: "enabled periodic_report requires profile_preset and route_ref" ) if route_ref.startswith("managed-session:") and managed_session_report_destination(value) is None: - raise ValueError("a managed Session report requires weekly-progress and its exact original Session route") + raise ValueError("a managed Session calendar requires weekly-progress or session-task and its exact original Session route") normalized = { "enabled": enabled, "profile_preset": profile_preset or None, "route_ref": route_ref or None, "timezone": timezone, } + if profile_preset == "session-task": + if managed_session_report_destination(value) is None: + raise ValueError("a scheduled task requires its exact original managed Session") + normalized["task"] = normalize_session_task(value.get("task")) + elif "task" in value: + raise ValueError("task instructions require the explicit session-task profile") schedule = normalize_report_cadence(value.get("schedule")) + if profile_preset == "session-task" and schedule is None: + raise ValueError("session-task requires its explicit calendar schedule") if schedule is not None: if schedule["timezone"] != timezone: raise ValueError("schedule.timezone must match periodic_report.timezone") @@ -112,4 +136,5 @@ def apply_change( "configuration_summary", "normalize_change", "normalize_configuration", + "normalize_session_task", ] diff --git a/loopx/capabilities/periodic_report/machine_defaults.py b/loopx/capabilities/periodic_report/machine_defaults.py index 729c681055..4d2c429f62 100644 --- a/loopx/capabilities/periodic_report/machine_defaults.py +++ b/loopx/capabilities/periodic_report/machine_defaults.py @@ -287,6 +287,13 @@ def _normalized_goal_subscription( "route_ref": route_ref, "timezone": timezone_name, } + if profile_preset == "session-task" or "task" in config: + from .goal_configuration import normalize_configuration + # A machine default must never enroll a Goal into executable work. + if source != "goal_override": + raise ValueError("session tasks require an explicit Goal override") + normalized = normalize_configuration(config) + subscription["task"] = normalized["task"] schedule = normalize_report_cadence(config.get("schedule")) if schedule is not None: if schedule["timezone"] != timezone_name: diff --git a/loopx/chat_goal_configuration_api.py b/loopx/chat_goal_configuration_api.py index 7596058589..681682a18e 100644 --- a/loopx/chat_goal_configuration_api.py +++ b/loopx/chat_goal_configuration_api.py @@ -226,7 +226,7 @@ def _goal_capability_options( "local_authority_shadow": {"enabled"}, "coordination_runtime_shadow": {"enabled"}, "lark_kanban_heartbeat_sync": {"enabled"}, - "periodic_report": {"enabled", "profile_preset", "route_ref", "timezone", "schedule"}, + "periodic_report": {"enabled", "profile_preset", "route_ref", "timezone", "schedule", "task"}, "reward_memory": {"config_path", "enabled_agents", "agent_id", "enabled"}, } if capability_id not in allowed: diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 6dc9539c42..cdb7f136e2 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -67,11 +67,13 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { } const observationOnly = request.work_kind === "monitor"; const reportWork = request.work_kind === "report"; + const scheduledWork = request.work_kind === "scheduled_task"; const requestFields = ["operation", "host", "actor", "turn_instance_id", "todo_id", "validation_command", "work_kind", ...(observationOnly ? ["monitor_deadline", "monitor_lease_version"] : []), - ...(reportWork ? ["report_window_id", "report_subscription_revision"] : [])]; + ...(reportWork ? ["report_window_id", "report_subscription_revision"] : []), + ...(scheduledWork ? ["schedule_window_id", "schedule_subscription_revision"] : [])]; if (Object.keys(request).some(key => !requestFields.includes(key)) - || request.operation !== "native_turn" || request.host !== "codex-app-server" || !["native_turn", "monitor", "report"].includes(String(request.work_kind)) + || request.operation !== "native_turn" || request.host !== "codex-app-server" || !["native_turn", "monitor", "report", "scheduled_task"].includes(String(request.work_kind)) || request.validation_command !== null || typeof request.turn_instance_id !== "string" || !/^[A-Za-z0-9._:-]{1,160}$/.test(request.turn_instance_id) || request.todo_id !== null && (typeof request.todo_id !== "string" || !/^todo_[A-Za-z0-9_-]{1,100}$/.test(request.todo_id))) { @@ -108,6 +110,42 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { } else if (params.report_settlement_recovery != null) { throw new EffectRuntimeRequestError("report recovery belongs only to its original calendar report"); } + if (scheduledWork) { + const basis = requireJsonObject(params.scheduled_task, "original scheduled task"); + const subscription = requireJsonObject(basis.subscription, "current task subscription"); + const window = requireJsonObject(basis.window, "original calendar window"); + const todo = requireJsonObject(basis.todo, "original finite task"); + const frozen = requireJsonObject(basis.definition, "original task definition"); + const originalActor = requireJsonObject(frozen.actor, "original task actor"); + const task = requireJsonObject(subscription.task, "configured task"); + const frozenTask = requireJsonObject(frozen.task, "frozen task"); + const sourceTask = requireJsonObject(basis.source_task, "current completed source Task"); + const originalSource = requireJsonObject(frozen.source_task, "original source Task"); + const inherited = ["action_kind", "task_domain", "task_repository", "capability_binding_ref", "priority", + "required_write_scopes", "required_capabilities", "target_capabilities", "required_decision_scopes", "excluded_agents"]; + const originalFields = [...inherited, "todo_id", "text", "claimed_by", "status", "task_class", + "completion_validation_sha256", "completion_validation_revision", "completion_turn_key"]; + if (subscription.enabled !== true || subscription.source !== "goal_override" + || subscription.profile_preset !== "session-task" || subscription.route_ref !== `managed-session:${session.session_id}` + || subscription.effective_revision !== request.schedule_subscription_revision + || window.subscription_revision !== request.schedule_subscription_revision + || window.window_id !== request.schedule_window_id || window.goal_id !== actor.goal_id || window.agent_id !== actor.agent_id + || fields.some(key => originalActor[key] !== actor[key]) + || frozenTask.source_todo_id !== task.source_todo_id || task.source_todo_id !== sourceTask.todo_id + || originalFields.some(key => JSON.stringify(sourceTask[key] ?? null) !== JSON.stringify(originalSource[key] ?? null)) + || sourceTask.claimed_by !== actor.agent_id || sourceTask.status !== "done" + || sourceTask.task_class !== "advancement_task" || sourceTask.completion_validation_required !== true + || sourceTask.action_kind === "generate_periodic_report" || /^cadence_[a-f0-9]{64}$/.test(String(sourceTask.target_key ?? "")) + || todo.todo_id !== request.todo_id || todo.target_key !== window.window_id + || todo.claimed_by !== actor.agent_id || todo.task_class !== "advancement_task" + || todo.status !== "open" || todo.text !== frozen.todo_text + || ["action_kind", "task_domain", "task_repository", "capability_binding_ref", "priority"].some(key => todo[key] !== sourceTask[key]) + || ["required_write_scopes", "required_capabilities", "target_capabilities", "required_decision_scopes", "excluded_agents"] + .some(key => JSON.stringify(todo[key] ?? []) !== JSON.stringify(sourceTask[key] ?? [])) + || todo.completion_validation_required !== true) { + throw new EffectRuntimeRequestError("scheduled work requires its original explicit task, validation, Session and calendar window"); + } + } if (turn ? turn.session_id !== session.session_id || turn.turn_id !== session.active_turn_id || !["starting", "running"].includes(String(turn.status)) || turn.origin !== "native_loopx" || turn.loopx_execution !== true || JSON.stringify(turn.loopx_request) !== JSON.stringify(request) diff --git a/tests/test_managed_session_calendar_tasks.py b/tests/test_managed_session_calendar_tasks.py new file mode 100644 index 0000000000..d36bde72e8 --- /dev/null +++ b/tests/test_managed_session_calendar_tasks.py @@ -0,0 +1,488 @@ +"""Real native File/SQLite calendar -> Task -> Turn -> original Core result. + +The model is a synthetic app-server that computes an independently checked +artifact. These cases do not qualify live-model understanding or production +deadlines, and do not enroll any user's existing conversation. +""" +from contextlib import contextmanager +from datetime import datetime, timedelta, timezone +import http.client +import json +from pathlib import Path +import sys +import threading +from types import SimpleNamespace + +import pytest + +from test_native_steward_private import steward as steward_fixture, finish +from test_native_steward_turn_host import managed_fixture, cli +from loopx.capabilities.native_chat.steward_monitor import pump_steward_monitors +from loopx.capabilities.native_chat.session_calendar_tasks import SessionCalendarTasks +from loopx.capabilities.periodic_report.cadence_journal import read_cadence_journal +from loopx.capabilities.periodic_report.goal_configuration import normalize_configuration +from loopx.agent_registry import load_goal_from_registry + +steward = steward_fixture + + +@contextmanager +def settings_server(steward): + from loopx.chat_server import ChatHTTPServer, ChatRequestHandler + store, runtime, _, transport, *_ = steward + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.chat_store, server.runtime_controller = store, runtime + server.registry_path, server.runtime_root_override = runtime.registry_path, store.root.parent + server.lark_private_conversations = transport + server.lark_goal_topic_runtime = SimpleNamespace(health_snapshot=lambda: {}, close=lambda: None) + server.verbose = False + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + def request(path, body, origin=None): + connection = http.client.HTTPConnection(*server.server_address, timeout=20) + try: + headers = {"Content-Type": "application/json"} + if origin: + headers["Origin"] = origin + connection.request("POST", path, json.dumps(body), headers=headers) + response = connection.getresponse() + return response.status, json.loads(response.read()) + finally: + connection.close() + try: + yield request + finally: + server.shutdown() + # This fixture lends the existing runtime to HTTP; the outer steward + # fixture owns its lifetime and continues exercising the real pump. + del server.runtime_controller + server.server_close() + thread.join(timeout=5) + + +def apply_configuration(request, goal_id, config): + body = {"goal_id": goal_id, "capability_id": "periodic_report", "configuration": config} + status, preview = request("/api/chat/goal-configuration/preview", body) + assert status == 201, preview + status, applied = request("/api/chat/goal-configuration/apply", + {**body, "expected_plan_revision": preview["plan_revision"]}) + assert status == 200, applied + return applied + + +def setup_task(steward, tmp_path, monkeypatch, provider, *, decision_scopes=()): + _, session, actor, unrelated, _, capture = managed_fixture(steward, tmp_path, monkeypatch, provider) + _, runtime, *_ = steward + workspace = Path(load_goal_from_registry(runtime.registry_path, actor["goal_id"])["repo"]) + (workspace / "numbers.json").write_text("[19, 23]") + script = tmp_path / "calendar-model.py" + script.write_text('''import json +from pathlib import Path +def respond(): + values = json.loads(Path("numbers.json").read_text()) + Path("total.json").write_text(json.dumps({"total": sum(values)})) + return json.dumps({"result_kind": "validated_completion", "classification": "implementation", + "summary": "The independently validated total is 42.", "next_action": "Keep future independent periods scheduled.", + "delivery_batch_scale": "implementation", "delivery_outcome": "outcome_progress", + "vision_unchanged_reason": "The original task and its scope remain active.", + "work_evidence": ["total.json computed from numbers.json"], + "validation_evidence": ["The original independent argv checks the computed artifact"]}) +''') + fake = Path(runtime.codex_bin) + # managed_fixture already installed its own synthetic result. Replace the + # complete string value without deriving the expected artifact from it. + source = fake.read_text() + import re + source, count = re.subn(r'"message": \'\{"result_kind".*?\}\'', + '"message": __import__("runpy").run_path(' + repr(str(script)) + ')["respond"]()', source) + assert count >= 1 + fake.write_text(source) + runtime.adapters.pop(session["session_id"], None) + validator = [sys.executable, "-c", "import json; from pathlib import Path; assert json.loads(Path('total.json').read_text()) == {'total': 42}"] + for scope in decision_scopes: + gate = cli(runtime, "todo", "add", "--goal-id", actor["goal_id"], "--role", "user", + "--text", "Approve this synthetic task's recurring direction.", "--task-class", "user_gate", + "--blocks-agent", actor["agent_id"], "--decision-scope", scope, "--execute") + cli(runtime, "todo", "complete", "--goal-id", actor["goal_id"], "--role", "user", + "--todo-id", gate["todo_id"], "--agent-id", actor["agent_id"], "--decision-outcome", "approve", + "--evidence", "Synthetic owner standing approval for this exact task scope.") + source_task = cli(runtime, "todo", "add", "--goal-id", actor["goal_id"], "--role", "agent", + "--text", "Read numbers.json, compute its sum, and save total.json. Return the result here.", + "--claimed-by", actor["agent_id"], "--task-class", "advancement_task", "--action-kind", "calculate_total", + "--validation-command-json", json.dumps(validator), + *(value for scope in decision_scopes for value in ("--required-decision-scope", scope)), "--execute") + # The user is scheduling a repeat of an existing completed native task. + # Calendar configuration cannot create a new host validator or silently + # resume unrelated deferred/blocked work. + from loopx.capabilities.native_chat.steward_managed_mode import ManagedSessionTurn + first, created = ManagedSessionTurn(runtime).submit(session["session_id"], agent_id=actor["agent_id"], + operation_id="completed-source-task", todo_id=source_task["todo_id"]) + assert created + completed = finish(runtime, first) + assert completed["status"] == "completed" and completed["response"]["native_loopx"]["status"] == "committed", completed + config = {"enabled": True, "profile_preset": "session-task", + "route_ref": "managed-session:" + session["session_id"], "timezone": "UTC", + "schedule": {"schema_version": "periodic_report_schedule_v0", "schedule_id": "ordinary-task", + "timezone": "UTC", "rrule": "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", "start_date": "2020-01-01"}, + "task": {"source_todo_id": source_task["todo_id"]}} + return session, actor, unrelated, capture, workspace, config + + +def fixed_clock(monkeypatch): + import loopx.capabilities.native_chat.session_calendar_tasks as module + point = [datetime(2020, 1, 1, 9, 0, 1, tzinfo=timezone.utc)] + class Clock(datetime): + @classmethod + def now(cls, tz=None): + return point[0].astimezone(tz) + monkeypatch.setattr(module, "datetime", Clock) + return point + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_calendar_preserves_canonical_decision_scopes_through_original_return(steward, tmp_path, monkeypatch, provider): + from loopx.control_plane.todos.contract import normalize_todo_required_decision_scopes + from loopx.capabilities.native_chat.steward_monitor import StewardMonitor + from loopx.capabilities.periodic_report.private_generation import attempt_directory + + scopes = ("direction:goal:repeat_task", "other:goal:review_total") + expected = normalize_todo_required_decision_scopes(scopes) + session, actor, unrelated, capture, _, config = setup_task( + steward, tmp_path, monkeypatch, provider, decision_scopes=scopes) + store, runtime, *_ = steward + sid = session["session_id"] + service = SessionCalendarTasks(runtime) + source, _ = service.source_task(actor, config["task"]) + assert source["status"] == "done" and source["required_decision_scopes"] == expected + history = store.messages(sid) + fixed_clock(monkeypatch) + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + turns = pump_steward_monitors(runtime) + assert len(turns) == 1, service.health(sid) + completed = finish(runtime, turns[0]) + assert completed["status"] == "completed", completed + assert completed["response"]["native_loopx"]["status"] == "committed", completed + rows = StewardMonitor(runtime).todos(completed["loopx_request"]["actor"]) + child = next(row for row in rows if row["todo_id"] == completed["loopx_request"]["todo_id"]) + assert child["status"] == "done" and child["required_decision_scopes"] == expected + assert next(row for row in rows if row["todo_id"] == unrelated)["status"] == "open" + returned = [row for row in store.messages(sid) if row.get("turn_id") == completed["turn_id"] and row["role"] == "agent"] + assert len(returned) == 1 and returned[0]["text"] == completed["response"]["message"] + assert "42" in returned[0]["text"] and store.messages(sid)[:len(history)] == history + after = capture.read_bytes() + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert capture.read_bytes() == after + journal = read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) + assert journal["task_completion"]["message_id"] == returned[0]["message_id"] + frozen = json.loads((attempt_directory(runtime.coordination_runtime_root, journal["window"]) / "task.json").read_text()) + assert frozen["source_task"]["required_decision_scopes"] == expected + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_calendar_executes_ordinary_task_and_returns_original_result(steward, tmp_path, monkeypatch, provider): + import loopx.capabilities.native_chat.session_calendar_tasks as module + from loopx.capabilities.native_chat.steward_managed_mode import ManagedSessionTurn + from loopx.capabilities.native_chat.steward_monitor import StewardMonitor + session, actor, unrelated, capture, workspace, config = setup_task(steward, tmp_path, monkeypatch, provider) + store, runtime, bot, *_ = steward + sid = session["session_id"] + point = datetime(2020, 1, 1, 8, 59, 59, tzinfo=timezone.utc) + class Clock(datetime): + @classmethod + def now(cls, tz=None): + return point.astimezone(tz) + monkeypatch.setattr(module, "datetime", Clock) + original_frontier = module.SessionCalendarTasks.frontier + def checked_frontier(service, session_id): + try: + return original_frontier(service, session_id) + except (OSError, ValueError, KeyError, TypeError, RuntimeError) as exc: + pytest.fail(f"The real calendar producer rejected its fixture: {exc}") + monkeypatch.setattr(module.SessionCalendarTasks, "frontier", checked_frontier) + history, calls, writes = store.messages(sid), capture.read_bytes(), list(bot.writes) + assert "scheduled_task" not in ManagedSessionTurn(runtime).inspect(sid) + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + assert ManagedSessionTurn(runtime).inspect(sid)["scheduled_task"]["configured"] + assert capture.read_bytes() == calls + assert not pump_steward_monitors(runtime) + assert read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) is None + point += timedelta(seconds=2) + runtime._steward_monitor_next_check.clear() + turns = pump_steward_monitors(runtime) + assert len(turns) == 1 + completed = finish(runtime, turns[0]) + assert completed["status"] == "completed", completed + assert completed["response"]["native_loopx"]["status"] == "committed", completed + assert completed["response"]["native_loopx"]["effects"]["quota_spent"] + assert json.loads((workspace / "total.json").read_text()) == {"total": 42} + assert store.messages(sid)[:len(history)] == history + returned = [row for row in store.messages(sid) if row.get("turn_id") == completed["turn_id"] and row["role"] == "agent"] + assert len(returned) == 1 and "42" in returned[0]["text"] + rows = StewardMonitor(runtime).todos(completed["loopx_request"]["actor"]) + assert next(row for row in rows if row["todo_id"] == completed["loopx_request"]["todo_id"])["status"] == "done" + assert next(row for row in rows if row["todo_id"] == unrelated)["status"] == "open" + recursive = {**config, "task": {"source_todo_id": completed["loopx_request"]["todo_id"]}} + assert request("/api/chat/goal-configuration/preview", {"goal_id": actor["goal_id"], + "capability_id": "periodic_report", "configuration": recursive})[0] == 400 + after = capture.read_bytes() + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + journal = read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) + assert journal["task_completion"]["message_id"] == returned[0]["message_id"] + assert journal["publication"] is None + assert capture.read_bytes() == after and bot.writes == writes + point += timedelta(days=1) + runtime._steward_monitor_next_check.clear() + second = pump_steward_monitors(runtime) + assert len(second) == 1 and second[0]["turn_id"] != completed["turn_id"] + assert finish(runtime, second[0])["response"]["native_loopx"]["status"] == "committed" + + +@pytest.mark.parametrize("fault", ["wrong-route", "missing-template", "report-profile", "injected-command", "no-calendar"]) +def test_task_declarations_never_enroll_other_routes_or_implicit_commands(fault): + config = {"enabled": True, "profile_preset": "session-task", "route_ref": "managed-session:" + "a" * 32, + "timezone": "UTC", "task": {"source_todo_id": "todo_original"}, + "schedule": {"schema_version": "periodic_report_schedule_v0", "schedule_id": "ordinary-task", + "timezone": "UTC", "rrule": "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", "start_date": "2020-01-01"}} + if fault == "wrong-route": + config["route_ref"] = "external-channel" + elif fault == "missing-template": + config["task"].pop("source_todo_id") + elif fault == "report-profile": + config["profile_preset"] = "weekly-progress" + elif fault == "injected-command": + config["task"]["validation_command"] = ["untrusted-command"] + else: + config.pop("schedule") + with pytest.raises((ValueError, TypeError)): + normalize_configuration(config) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_known_failed_repeat_is_retained_without_replay_then_explicitly_cancelled(steward, tmp_path, monkeypatch, provider): + session, actor, _, capture, workspace, config = setup_task(steward, tmp_path, monkeypatch, provider) + point = fixed_clock(monkeypatch) + store, runtime, bot, *_ = steward + sid = session["session_id"] + # The synthetic model claims success; its actual artifact is wrong. The + # independent native validator must hold completion and keep that failure. + (workspace / "numbers.json").write_text("[1, 2]") + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + failed = finish(runtime, pump_steward_monitors(runtime)[0]) + assert failed["status"] == "completed" and failed["response"]["native_loopx"]["status"] == "failed", failed + assert json.loads((workspace / "total.json").read_text()) == {"total": 3} + calls, writes, messages = capture.read_bytes(), list(bot.writes), store.messages(sid) + for _ in range(2): + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert capture.read_bytes() == calls and bot.writes == writes and store.messages(sid) == messages + apply_configuration(request, actor["goal_id"], None) + journal = read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) + assert journal["retirement"]["reason"] == "owner_cancelled_failed_task" and journal["publication"] is None + apply_configuration(request, actor["goal_id"], config) + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) # Re-enable cannot replay this failed period. + (workspace / "numbers.json").write_text("[19, 23]") + point[0] += timedelta(days=1) + runtime._steward_monitor_next_check.clear() + completed = finish(runtime, pump_steward_monitors(runtime)[0]) + assert completed["turn_id"] != failed["turn_id"] + assert completed["response"]["native_loopx"]["status"] == "committed", completed + assert store.load_turn(sid, failed["turn_id"]) == failed + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_real_settings_only_select_existing_completed_source_and_current_preview(steward, tmp_path, monkeypatch, provider): + session, actor, incomplete, capture, _, config = setup_task(steward, tmp_path, monkeypatch, provider) + _, runtime, bot, *_ = steward + calls, writes = capture.read_bytes(), list(bot.writes) + with settings_server(steward) as request: + body = {"goal_id": actor["goal_id"], "capability_id": "periodic_report", "configuration": config} + injected = {**config, "task": {**config["task"], "validation_command": ["untrusted-command"]}} + assert request("/api/chat/goal-configuration/preview", {**body, "configuration": injected})[0] in (400, 409) + for source in [incomplete, "todo_missing"]: + denied = {**config, "task": {"source_todo_id": source}} + assert request("/api/chat/goal-configuration/preview", {**body, "configuration": denied})[0] == 400 + assert request("/api/chat/goal-configuration/preview", body, "https://foreign.example")[0] == 403 + status, preview = request("/api/chat/goal-configuration/preview", body) + assert status == 201 + assert request("/api/chat/goal-configuration/apply", {**body, + "configuration": {**config, "enabled": False}, "expected_plan_revision": preview["plan_revision"]})[0] == 409 + assert capture.read_bytes() == calls and bot.writes == writes + assert not SessionCalendarTasks(runtime).health(session["session_id"])["configured"] + + +@pytest.mark.parametrize("fault", ["queued", "busy", "paused", "unregistered", "stopped"]) +def test_original_user_work_and_withdrawn_authority_win_without_model(steward, tmp_path, monkeypatch, fault): + session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, "sqlite") + fixed_clock(monkeypatch) + store, runtime, bot, *_ = steward + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], {**config, "enabled": fault != "paused"}) + if fault == "queued": + store.create_queued_turn(session["session_id"], client_turn_id="owner-first", message="Retain this user work first", origin="web") + elif fault == "busy": + store.accept_managed_turn(session["session_id"], client_turn_id="owner-active", message="Original accepted user work", origin="web") + elif fault in {"unregistered", "stopped"}: + registry = json.loads(runtime.registry_path.read_text()) + goal = next(row for row in registry["goals"] if row["id"] == actor["goal_id"]) + if fault == "stopped": + goal["status"] = "stopped" + else: + goal["coordination"]["registered_agents"] = ["peer"] + runtime.registry_path.write_text(json.dumps(registry)) + calls, writes, messages = capture.read_bytes(), list(bot.writes), store.messages(session["session_id"]) + assert not runtime.closed.is_set() + assert not pump_steward_monitors(runtime) + assert capture.read_bytes() == calls and bot.writes == writes and store.messages(session["session_id"]) == messages + assert read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) is None + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_completed_task_and_original_body_reconcile_after_runtime_restart(steward, tmp_path, monkeypatch, provider): + from loopx.chat_runtime import ChatRuntimeController + from loopx.chat_store import ChatSessionStore + import loopx.capabilities.native_chat.session_calendar_tasks as module + session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, provider) + fixed_clock(monkeypatch) + store, runtime, bot, *_ = steward + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + completed = finish(runtime, pump_steward_monitors(runtime)[0]) + assert completed["response"]["native_loopx"]["status"] == "committed" + original = module.record_cadence_task_completion + def lost_response(**kwargs): + original(**kwargs) + raise OSError("synthetic lost completion acknowledgement") + monkeypatch.setattr(module, "record_cadence_task_completion", lost_response) + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + monkeypatch.setattr(module, "record_cadence_task_completion", original) + runtime.close() + before = capture.read_bytes(), store.messages(session["session_id"]), list(bot.writes) + recovered = ChatRuntimeController(store=ChatSessionStore(store.root.parent), codex_bin=runtime.codex_bin, + project_contexts=runtime.project_contexts, registry_path=runtime.registry_path) + try: + assert SessionCalendarTasks(recovered).frontier(session["session_id"]) == (None, False) + assert SessionCalendarTasks(recovered).health(session["session_id"])["completion_verified"] + assert (capture.read_bytes(), recovered.store.messages(session["session_id"]), list(bot.writes)) == before + finally: + recovered.close() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_archived_source_and_completed_period_remain_usable_without_replay(steward, tmp_path, monkeypatch, provider): + session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, provider) + point = fixed_clock(monkeypatch) + _, runtime, *_ = steward + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + first = finish(runtime, pump_steward_monitors(runtime)[0]) + assert first["response"]["native_loopx"]["status"] == "committed" + cli(runtime, "todo", "archive-completed", "--goal-id", actor["goal_id"], "--role", "agent", + "--max-active-done", "0", "--execute") + calls = capture.read_bytes() + service = SessionCalendarTasks(runtime) + assert service.frontier(session["session_id"]) == (None, False) + assert service.health(session["session_id"])["completion_verified"] and capture.read_bytes() == calls + point[0] += timedelta(days=1) + turn, created = service.frontier(session["session_id"]) + assert created and turn["turn_id"] != first["turn_id"] + assert finish(runtime, turn)["response"]["native_loopx"]["status"] == "committed" + + +@pytest.mark.parametrize("fault", ["text", "scope", "paused", "ordinary-entry"]) +def test_frozen_task_cannot_change_work_or_scope_before_dispatch(steward, tmp_path, monkeypatch, fault): + from loopx.capabilities.native_chat.steward_managed_mode import ManagedSessionTurn + session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, "sqlite") + fixed_clock(monkeypatch) + _, runtime, bot, *_ = steward + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + service = SessionCalendarTasks(runtime) + with monkeypatch.context() as before_launch: + before_launch.setattr(ManagedSessionTurn, "submit", lambda *_a, **_kw: (None, False)) + assert service.frontier(session["session_id"]) == (None, False) + window = read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"])["window"] + todo = service._todo(service.selected(session["session_id"])["actor"], window) + if fault in {"text", "scope"}: + cli(runtime, "todo", "update", "--goal-id", actor["goal_id"], "--todo-id", todo["todo_id"], + "--agent-id", actor["agent_id"], "--update-operation-id", "change-frozen-task", + *(["--text", "Run a different task"] if fault == "text" else ["--required-capability", "filesystem"])) + else: + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], {**config, "enabled": False}) + calls, writes = capture.read_bytes(), list(bot.writes) + if fault == "ordinary-entry": + accepted, created = ManagedSessionTurn(runtime).submit(session["session_id"], agent_id=actor["agent_id"], + operation_id="ordinary-bypass", todo_id=todo["todo_id"]) + assert created + rejected = finish(runtime, accepted) + assert rejected["response"]["native_loopx"]["status"] == "failed", rejected + else: + with pytest.raises(ValueError, match="subscription changed" if fault == "paused" else "scheduled work requires"): + ManagedSessionTurn(runtime).submit(session["session_id"], agent_id=actor["agent_id"], + operation_id="task." + window["window_id"], todo_id=todo["todo_id"], scheduled_window=window) + assert capture.read_bytes() == calls and bot.writes == writes + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_unknown_postcommit_return_never_becomes_cancelled_failure_or_model_replay(steward, tmp_path, monkeypatch, provider): + session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, provider) + fixed_clock(monkeypatch) + store, runtime, bot, *_ = steward + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + append = store.append_event + lost = [] + def lose_native_return(*args, **kwargs): + result = append(*args, **kwargs) + if kwargs.get("kind") == "native_loopx.result" and not lost: + lost.append(True) + raise OSError("synthetic lost return after the original native commit") + return result + with monkeypatch.context() as lost_return: + lost_return.setattr(store, "append_event", lose_native_return) + failed = finish(runtime, pump_steward_monitors(runtime)[0]) + assert failed["status"] == "failed" and lost + before = capture.read_bytes(), store.messages(session["session_id"]), list(bot.writes) + for _ in range(2): + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert not SessionCalendarTasks(runtime).health(session["session_id"])["completion_verified"] + body = {"goal_id": actor["goal_id"], "capability_id": "periodic_report", "configuration": None} + assert request("/api/chat/goal-configuration/preview", body)[0] == 400 + assert load_goal_from_registry(runtime.registry_path, actor["goal_id"])["control_plane"]["periodic_report"] == config + assert (capture.read_bytes(), store.messages(session["session_id"]), list(bot.writes)) == before + assert store.load_turn(session["session_id"], failed["turn_id"]) == failed + + +def test_partial_progress_keeps_period_open_without_an_automatic_retry(steward, tmp_path, monkeypatch): + session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, "sqlite") + fixed_clock(monkeypatch) + _, runtime, *_ = steward + script = tmp_path / "calendar-model.py" + script.write_text(script.read_text().replace('"result_kind": "validated_completion"', '"result_kind": "validated_progress"')) + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + completed = finish(runtime, pump_steward_monitors(runtime)[0]) + # This native Task requires declared completion before accountable + # writeback. Partial model work is retained at that gate, not treated + # as a completed calendar period or permission to invoke it again. + native = completed["response"]["native_loopx"] + assert native["ok"] is False and native.get("status") != "committed", completed + assert native["journal_observation"]["recorded_effects"]["quota_spent"] is False + service = SessionCalendarTasks(runtime) + assert not service.health(session["session_id"])["completion_verified"] + calls = capture.read_bytes() + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) and capture.read_bytes() == calls + journal = read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) + assert not journal.get("task_completion") and journal["publication"] is None + assert service._todo(service.selected(session["session_id"])["actor"], journal["window"])["status"] == "open" + assert request("/api/chat/goal-configuration/preview", {"goal_id": actor["goal_id"], + "capability_id": "periodic_report", "configuration": None})[0] == 400 diff --git a/tests/test_native_steward_monitor.py b/tests/test_native_steward_monitor.py index 739e18e06f..50a519329c 100644 --- a/tests/test_native_steward_monitor.py +++ b/tests/test_native_steward_monitor.py @@ -121,11 +121,7 @@ def record(**kw): # time. Advance to the next real scheduler tick before authoring this # synthetic second window; do not retry a rejected/unknown operation. from loopx.control_plane.runtime.time import now_utc - checked = datetime.fromisoformat(after["last_checked_at"].replace("Z", "+00:00")) - limit = time.monotonic() + 2 - while now_utc() <= checked: - assert time.monotonic() < limit - time.sleep(.02) + wait_for_next_observation_tick(after) edit_local_watch(restarted, actor, todo_id, "--next-due-at", now_utc().isoformat()) second = pump_steward_monitors(restarted) @@ -401,6 +397,18 @@ def test_unreadable_conversation_does_not_starve_another_native_monitor(steward, assert capture.read_bytes() == calls +def wait_for_next_observation_tick(todo): + """A new keyed effect needs a newer observation, not only another due ID.""" + from loopx.control_plane.runtime.time import now_utc + if not todo.get("last_checked_at"): + return + checked = datetime.fromisoformat(todo["last_checked_at"].replace("Z", "+00:00")) + limit = time.monotonic() + 2 + while now_utc() <= checked: + assert time.monotonic() < limit, "the native observation clock did not advance" + time.sleep(.02) + + def due_window(monitor, actor, deadline="2001-01-01T00:00:00Z"): """Advance a synthetic window through the real native Todo/lease owners. @@ -409,6 +417,7 @@ def due_window(monitor, actor, deadline="2001-01-01T00:00:00Z"): from uuid import uuid4 from loopx.capabilities.native_chat.steward_commands import native_command todo = monitor.current(actor) + wait_for_next_observation_tick(todo) key = "test-window." + uuid4().hex lease = monitor._acquire(actor, todo, key)["lease"] try: