diff --git a/docs/assets/personal-workspace/steward-reports/session-failed-cancel-before.png b/docs/assets/personal-workspace/steward-reports/session-failed-cancel-before.png new file mode 100644 index 0000000000..e94762d3f0 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-failed-cancel-before.png differ diff --git a/docs/assets/personal-workspace/steward-reports/session-failed-cancel-cleared.png b/docs/assets/personal-workspace/steward-reports/session-failed-cancel-cleared.png new file mode 100644 index 0000000000..435898d664 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-failed-cancel-cleared.png differ diff --git a/docs/assets/personal-workspace/steward-reports/session-failed-cancel-narrow.png b/docs/assets/personal-workspace/steward-reports/session-failed-cancel-narrow.png new file mode 100644 index 0000000000..bb1f9f4ea2 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-failed-cancel-narrow.png differ diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index d19db650de..7ad2211799 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -581,9 +581,18 @@ window, failure and message history remain. Re-enable that exact plan to continue eligible native recovery. An unpublished frozen window prevents a schedule/destination change from stranding it. Failed/unknown generation is retained for repair by its original execution owner; toggling the setting does -not replay it. This slice does not add cancellation of an invalid authored -window, natural-language schedule authoring or arbitrary task profiles. Those -remain part of the existing general recurring-task work. +not replay it. After a known terminal generation failure, with no generated +report or delivery attempt, **Restore device defaults** can clear this Goal's +subscription through the existing preview/apply transaction. Verified apply +blocks the original failed Task and retires its window without changing its +artifacts, costs, Turn or publication cursor. Preview alone does not cancel. +If cancellation is interrupted after the configuration clears, retry the same +clear request: the original frozen Session identity and failure recover the +remaining retirement, without a model call or replay. Re-enabling the same +calendar does not retry that failed period; the next independent period can run. +Existing reports and unknown or partial delivery retain the original protection. +Natural-language schedule authoring and arbitrary task profiles remain part of +the existing general recurring-task work. The packaged journey uses disposable Core/SQLite state and synthetic model and provider transports. It checks future-date inactivity, preview/apply/reload, @@ -591,6 +600,18 @@ one automatic full-body return, retained failure and pause. The 390×844 view is viewport emulation; this evidence does not establish real-model report quality, production backlog completion or a natural production deadline. +The cancellation check also uses that packaged UI and real disposable SQLite +state: a synthetic invalid editorial result remains failed, clear preview leaves +it intact, and applied clear retires it without another model call, message or +provider send. File/SQLite regressions separately cover interrupted cancellation, +unchanged artifacts and history, unknown effects and the next independent period. + +| Retained failed period | Cleared Goal override, failure retained | +| --- | --- | +| ![Failed Session report before clear](../assets/personal-workspace/steward-reports/session-failed-cancel-before.png) | ![Cleared report subscription](../assets/personal-workspace/steward-reports/session-failed-cancel-cleared.png) | + +![Cleared subscription in a narrow viewport](../assets/personal-workspace/steward-reports/session-failed-cancel-narrow.png) + | Before: raw route | After: original Agent conversation and verified return | | --- | --- | | ![Previous report settings](../assets/personal-workspace/steward-reports/session-before-desktop.png) | ![Report returned to the selected original conversation](../assets/personal-workspace/steward-reports/session-delivered-desktop.png) | diff --git a/loopx/capabilities/native_chat/steward_reports.py b/loopx/capabilities/native_chat/steward_reports.py index 7bca77297f..6dfad818f0 100644 --- a/loopx/capabilities/native_chat/steward_reports.py +++ b/loopx/capabilities/native_chat/steward_reports.py @@ -83,10 +83,28 @@ def _known_generation_failure(self, actor: dict[str, Any], window: dict[str, Any request = (turn or {}).get("loopx_request") or {} native = ((turn or {}).get("response") or {}).get("native_loopx") or {} if (turn and turn.get("status") == "completed" and native.get("status") == "failed" - and request.get("actor") == actor and request.get("report_window_id") == window["window_id"]): + and request.get("actor") == actor and request.get("work_kind") == "report" + and request.get("report_window_id") == window["window_id"] + and request.get("report_subscription_revision") == window["subscription_revision"]): return {"turn_id": turn["turn_id"], "todo_id": request["todo_id"]} return None + def _cancel_failed_window(self, actor: dict[str, Any], window: dict[str, Any], + failure: dict[str, str], *, cadence_locked: bool = False) -> None: + 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 report 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.report." + window["window_id"], "--status", "blocked", + "--clear-resume-when", "--reason", + "Owner cancelled a known failed report window; retain the failed task, costs and evidence, not completed or delivered.", "--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"]) + def configure(self, binding_id: str, *, operation: str, schedule: dict[str, Any] | None = None, expected_plan_id: str | None = None, execute: bool = False) -> dict[str, Any]: from ...configure_goal import configure_goal @@ -140,14 +158,7 @@ def configure(self, binding_id: str, *, operation: str, schedule: dict[str, Any] if configuration_summary(load_goal_from_registry(self.controller.registry_path, actor["goal_id"])) != proposed: raise ValueError("native report configuration readback changed") if plan["retires_failed_window"]: - from ..periodic_report.cadence_journal import retire_failed_cadence_window - key = "cancel.report." + journal["window"]["window_id"] - native_command(self.controller, ["todo", "update", "--goal-id", actor["goal_id"], - "--todo-id", failure["todo_id"], "--agent-id", actor["agent_id"], - "--update-operation-id", key, "--status", "blocked", "--clear-resume-when", "--reason", - "Owner cancelled a known failed report window; retain the failed task, costs and evidence, not completed or delivered.", "--execute"]) - retire_failed_cadence_window(runtime_root=self.root, goal_id=actor["goal_id"], - expected_window_id=journal["window"]["window_id"], source_receipt_id=failure["turn_id"]) + self._cancel_failed_window(actor, journal["window"], failure) getattr(self.controller, "_steward_monitor_next_check", {}).pop(binding_id, None) receipt = {"ok": True, "applied": True, "plan": plan, "report": self.health(binding_id), "model_invoked": False, "external_writes_performed": False} @@ -344,13 +355,15 @@ def session_report_configuration_guard(controller: Any, goal_id: str, proposed: """Keep the existing settings transaction and native window in one ordering. Preparing source artifacts takes the same Goal-local calendar lock. A - changed plan cannot strand a frozen undelivered report; pause and restoring - that same plan remain possible. Other destinations keep their old path. + changed plan cannot strand a frozen undelivered report. Clearing a known + failed generation retires only that attempt after a verified settings apply; + previews and pauses retain it. Other destinations keep their old path. """ from ..periodic_report.goal_configuration import managed_session_report_destination from ..periodic_report.cadence_journal import cadence_journal_path current = configuration_summary(load_goal_from_registry(controller.registry_path, goal_id) or {}) or {} - if not managed_session_report_destination(current) and not managed_session_report_destination(proposed or {}): + if (not managed_session_report_destination(current) and not managed_session_report_destination(proposed or {}) + and (current or proposed is not None)): yield return with exclusive_file_lock(cadence_journal_path(controller.coordination_runtime_root, goal_id), @@ -362,6 +375,7 @@ def session_report_configuration_guard(controller: Any, goal_id: str, proposed: if actor["goal_id"] != goal_id: raise ValueError("select the original managed Agent Session in this Goal") 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) @@ -369,9 +383,20 @@ def session_report_configuration_guard(controller: Any, goal_id: str, proposed: 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"} - if before != after: - raise PendingStewardReportError("The original frozen report still requires verified delivery. Pause or restore the same plan; change its schedule or conversation after delivery.") - yield + if before != after or not current and proposed is None: + 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): + 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 + def after_verified_apply(): + if configuration_summary(load_goal_from_registry(controller.registry_path, goal_id)) is not None: + raise ValueError("the original report configuration was not cleared") + if reports._frozen_actor(cancellation[1]) != cancellation[0]: + raise ValueError("the original failed report actor changed before cancellation") + reports._cancel_failed_window(*cancellation, cadence_locked=True) + yield after_verified_apply if cancellation else None class ManagedSessionReports(StewardReports): @@ -458,16 +483,23 @@ def read(): frozen = prepare_session_report(root=self.root, window=window, actor=actor, read=read, reporting=reporting) return report_prompt(frozen, attempt_directory(self.root, window)) - def delivery_current(self, session_id: str, window: dict[str, Any]) -> bool: + def _frozen_actor(self, window: dict[str, Any]) -> dict[str, Any] | None: from ..periodic_report.pending_intent import _canonical_digest - if self.subscription(session_id, expected_revision=window["subscription_revision"]) is None: - return False - actor = self.selected(session_id)["actor"] - frozen = json.loads((attempt_directory(self.root, window) / "editorial-request.json").read_text(encoding="utf-8")) - return (actor == frozen.get("managed_session_actor") - and frozen.get("window") == window and actor["goal_id"] == window["goal_id"] - and actor["agent_id"] == window["agent_id"] - and frozen.get("request_digest") == _canonical_digest({k: v for k, v in frozen.items() if k != "request_digest"})) + path = attempt_directory(self.root, window) / "editorial-request.json" + if not path.is_file(): + return None + frozen = json.loads(path.read_text(encoding="utf-8")) + saved = frozen.get("managed_session_actor") + if not isinstance(saved, dict): + return None + actor = self.selected(saved.get("session_id"))["actor"] + return actor if (actor == saved and frozen.get("window") == window + and actor["goal_id"] == window["goal_id"] and actor["agent_id"] == window["agent_id"] + and frozen.get("request_digest") == _canonical_digest({k: v for k, v in frozen.items() if k != "request_digest"})) else None + + def delivery_current(self, session_id: str, window: dict[str, Any]) -> bool: + return (self.subscription(session_id, expected_revision=window["subscription_revision"]) is not None + and self._frozen_actor(window) == self.selected(session_id)["actor"]) def publish(self, session_id: str, window: dict[str, Any]) -> bool: """Read the original native commit, then publish/read back one Core message. diff --git a/loopx/capabilities/periodic_report/cadence_journal.py b/loopx/capabilities/periodic_report/cadence_journal.py index 6149915149..3a62115ffd 100644 --- a/loopx/capabilities/periodic_report/cadence_journal.py +++ b/loopx/capabilities/periodic_report/cadence_journal.py @@ -107,21 +107,28 @@ def retire_failed_cadence_window(*, runtime_root: Path, goal_id: str, This retires admission only; it neither retries a model nor drops evidence. """ path = cadence_journal_path(runtime_root, goal_id) - if not isinstance(source_receipt_id, str) or not source_receipt_id: - raise ValueError("failed cadence retirement requires its original receipt") with exclusive_file_lock(path, policy=LockAcquisitionPolicy.MUTATION, operation="retire_failed_report_window"): - 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: - raise ValueError("the unpublished failed report window changed") - if current.get("retirement"): - if current["retirement"]["source_receipt_id"] != source_receipt_id: - raise ValueError("the failed window retirement changed its source") - return current - current["retirement"] = {"reason": "owner_cancelled_failed_generation", - "source_receipt_id": source_receipt_id, "retired_at": datetime.now(timezone.utc).isoformat()} - atomic_write_json(path, current) + 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) + + +def _retire_failed_cadence_window_locked(*, runtime_root: Path, goal_id: str, + expected_window_id: str, source_receipt_id: str) -> 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: + 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: + raise ValueError("the unpublished failed report window changed") + if current.get("retirement"): + if current["retirement"]["source_receipt_id"] != source_receipt_id: + raise ValueError("the failed window retirement changed its source") return current + current["retirement"] = {"reason": "owner_cancelled_failed_generation", + "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 def _archive_predecessor(path: Path, previous: dict[str, Any]) -> str: diff --git a/loopx/chat_goal_configuration_api.py b/loopx/chat_goal_configuration_api.py index 707491a369..7596058589 100644 --- a/loopx/chat_goal_configuration_api.py +++ b/loopx/chat_goal_configuration_api.py @@ -617,6 +617,7 @@ def _send_applied_goal_configuration( desired_public: Mapping[str, Any], goal_id: str, plan: Mapping[str, Any], + after_verified_apply: Callable[[], None] | None = None, ) -> None: try: readback_public = self._read_public_goal_configuration(goal_id) @@ -644,6 +645,8 @@ def _send_applied_goal_configuration( ) return raise RuntimeError("Goal configuration readback did not verify") + if after_verified_apply is not None: + after_verified_apply() self._send_json( { "ok": True, @@ -674,7 +677,7 @@ def _goal_configuration_update(self, *, execute: bool) -> None: controller = getattr(self.server, "runtime_controller", None) guard = (session_report_configuration_guard(controller, goal_id, configuration) if capability_id == "periodic_report" and controller is not None else nullcontext()) - with guard: + with guard as after_verified_apply: current_public, desired_public, plan, desired_configuration = ( self._prepare_goal_configuration_plan( goal_id=goal_id, @@ -728,6 +731,7 @@ def _goal_configuration_update(self, *, execute: bool) -> None: desired_public=desired_public, goal_id=goal_id, plan=plan, + after_verified_apply=after_verified_apply, ) except (TypeError, ValueError) as exc: stale = "preview again" in str(exc) diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 0ba91fe825..e9a9265081 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -255,6 +255,143 @@ def test_local_report_configuration_retains_frozen_window_and_original_agent_ele goal_id=session["goal_id"], agent_id=session["managed_agent_id"], execute=True)["status"] == "original_conversation_required" +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("interruption", [None, "settings_cleared", "task_blocked"]) +def test_local_cancel_retires_only_a_known_failed_window_after_verified_apply(steward, tmp_path, monkeypatch, provider, interruption): + from datetime import timezone + from loopx.capabilities.native_chat.steward_monitor import StewardMonitor + from loopx.capabilities.native_chat.steward_reports import session_report_configuration_guard + from loopx.capabilities.periodic_report import cadence_runtime + from loopx.configure_goal import configure_goal + import loopx.capabilities.native_chat.steward_reports as report_host + session, reports, config, _ = configured_local_report(steward, tmp_path, monkeypatch, provider) + store, runtime, bot, _, _, capture, _ = steward + sid, goal_id = session["session_id"], session["goal_id"] + point = datetime.now(timezone.utc).replace(hour=9, minute=1, second=0, microsecond=0) - timedelta(days=2) + + class CalendarClock(datetime): + @classmethod + def now(cls, tz=None): + return point.astimezone(tz) if tz else point.replace(tzinfo=None) + monkeypatch.setattr(report_host, "datetime", CalendarClock) + monkeypatch.setattr(cadence_runtime, "datetime", CalendarClock) + install_model(steward, tmp_path, sid) + model = tmp_path / "report_model.py" + valid = model.read_text() + model.write_text(valid.replace('"request_digest": request["request_digest"]', '"request_digest": "known-invalid"')) + failed = finish(runtime, pump_steward_monitors(runtime)[0]) + assert failed["response"]["native_loopx"]["status"] == "failed" + journal = read_cadence_journal(runtime_root=reports.root, goal_id=goal_id) + folder = attempt_directory(reports.root, journal["window"]) + artifacts = {p.name: p.read_bytes() for p in folder.iterdir() if p.is_file()} + calls, writes, registry = capture.read_bytes(), list(bot.writes), runtime.registry_path.read_bytes() + assert not (folder / "private-generation.json").exists() + # Preview and an apply that fails before verified readback cannot cancel. + with session_report_configuration_guard(runtime, goal_id, None) as after_apply: + assert callable(after_apply) + with pytest.raises(ValueError, match="not cleared"): + after_apply() + with pytest.raises(OSError, match="write unavailable"): + with session_report_configuration_guard(runtime, goal_id, None): + raise OSError("Synthetic settings write unavailable") + assert read_cadence_journal(runtime_root=reports.root, goal_id=goal_id) == journal + assert runtime.registry_path.read_bytes() == registry + # Existing generation/unknown-delivery evidence must still prevent clear. + for name in ["private-generation.json", "private-delivery.json"]: + retained = folder / name + retained.write_text("{}") + with pytest.raises(ValueError, match="requires verified delivery"): + with session_report_configuration_guard(runtime, goal_id, None): + pytest.fail("an existing effect must not be cancelled as a generation failure") + retained.unlink() + if interruption: + from loopx.capabilities.periodic_report import cadence_journal + def unavailable_retirement(**_kwargs): + raise OSError("Synthetic cancellation receipt unavailable") + with pytest.raises(OSError, match="cancellation receipt unavailable"), monkeypatch.context() as interrupted: + if interruption == "task_blocked": + interrupted.setattr(cadence_journal, "_retire_failed_cadence_window_locked", unavailable_retirement) + with session_report_configuration_guard(runtime, goal_id, None) as after_apply: + assert configure_goal(registry_path=runtime.registry_path, goal_id=goal_id, + clear_periodic_report_configuration=True, execute=True)["ok"] + if interruption == "settings_cleared": + raise OSError("Synthetic cancellation receipt unavailable") + after_apply() + assert read_cadence_journal(runtime_root=reports.root, goal_id=goal_id) == journal + assert capture.read_bytes() == calls and bot.writes == writes + # A retry uses the original frozen actor even after configuration cleared. + with session_report_configuration_guard(runtime, goal_id, None) as after_apply: + assert callable(after_apply) + assert configure_goal(registry_path=runtime.registry_path, goal_id=goal_id, + clear_periodic_report_configuration=True, execute=True)["ok"] + after_apply() + retired = read_cadence_journal(runtime_root=reports.root, goal_id=goal_id) + assert retired["retirement"]["source_receipt_id"] == failed["turn_id"] + assert retired["publication"] is None + task = next(row for row in StewardMonitor(runtime).todos(failed["loopx_request"]["actor"]) + if row["todo_id"] == failed["loopx_request"]["todo_id"]) + assert task["status"] == "blocked" and not task.get("done") + assert capture.read_bytes() == calls and bot.writes == writes + assert store.load_turn(sid, failed["turn_id"]) == failed + assert {p.name: p.read_bytes() for p in folder.iterdir() if p.is_file()} == artifacts + assert read_periodic_report_publication_cursor(runtime_root=reports.root, goal_id=goal_id, + agent_id=session["managed_agent_id"]) is None + with session_report_configuration_guard(runtime, goal_id, None) as after_apply: + assert after_apply is None + # The old failed attempt stays retired; a new independent period works. + with session_report_configuration_guard(runtime, goal_id, config) as after_apply: + assert after_apply is None + assert configure_goal(registry_path=runtime.registry_path, goal_id=goal_id, + periodic_report_configuration=config, execute=True)["ok"] + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert capture.read_bytes() == calls + model.write_text(valid) + point += 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" + assert reports.health(sid)["publication_verified"] + assert store.load_turn(sid, failed["turn_id"]) == failed + assert bot.writes == writes + + +@pytest.mark.parametrize("phase,expected_status", [ + ("preview", 201), ("stale", 409), ("write_failure", 500), ("readback_failure", 500), ("applied", 200), +]) +def test_settings_retirement_runs_after_verified_apply_before_the_http_receipt(monkeypatch, phase, expected_status): + from contextlib import contextmanager + from test_chat_goal_configuration_api import _PeriodicClearHandler + import loopx.capabilities.native_chat.steward_reports as report_host + body = {"goal_id": "goal-example", "capability_id": "periodic_report", "configuration": None} + preview = _PeriodicClearHandler(body) + preview._goal_configuration_update(execute=False) + revision = preview.responses[0]["plan_revision"] + handler = _PeriodicClearHandler(body if phase == "preview" else + {**body, "expected_plan_revision": "stale" if phase == "stale" else revision}) + handler.server.runtime_controller = object() + calls = [] + + @contextmanager + def guarded_clear(_controller, goal_id, proposed): + assert goal_id == "goal-example" and proposed is None + def retire(): + assert not handler.override_present and not handler.responses + calls.append("retired") + yield retire + monkeypatch.setattr(report_host, "session_report_configuration_guard", guarded_clear) + if phase == "write_failure": + writer = handler._goal_configuration_writer() + monkeypatch.setattr(handler, "_goal_configuration_writer", lambda: lambda **kwargs: + {**writer(**kwargs), **({"ok": False} if kwargs["execute"] else {})}) + if phase == "readback_failure": + monkeypatch.setattr(handler, "_goal_configuration_readback_matches", lambda *_: False) + handler._goal_configuration_update(execute=phase != "preview") + assert handler.responses[0]["status_code"] == expected_status + assert calls == (["retired"] if phase == "applied" else []) + + def configured(steward, tmp_path, monkeypatch, provider="sqlite", start_date="2001-01-01"): isolate_sqlite_runtime(tmp_path, monkeypatch) store, runtime, _, transport, binding, capture, _ = steward