From f39c40c5988ac5e5c1c4832a789ae56d51039585 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:55:42 +0530 Subject: [PATCH 1/8] refactor(runtime): drive reconciliation inside existing live scopes Share cadence, step backoff and wakeup handling across native observation, manager pumps and Lark private delivery. Preserve live pools, futures, cursors and provider-owned authorization and health callbacks during step recovery. Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/extensions.md | 12 ++- .../capabilities/manager_context/roundtrip.py | 29 +++---- loopx/extensions/background_runtime.py | 32 ++++++++ .../lark/goal_topic_runtime_service.py | 16 ++-- .../loopx_codex_provider_routing/native.py | 29 ++++--- tests/extensions/test_background_runtime.py | 77 +++++++++++++++++++ .../test_native_observer_health.py | 5 +- tests/test_lark_private_conversations.py | 51 ++++++++++++ tests/test_manager_context_roundtrip.py | 8 ++ 9 files changed, 224 insertions(+), 35 deletions(-) diff --git a/docs/reference/extensions.md b/docs/reference/extensions.md index 8cc1388b5b..8c07eb62b4 100644 --- a/docs/reference/extensions.md +++ b/docs/reference/extensions.md @@ -25,7 +25,9 @@ LoopX Core Python-hosted background loops share `loopx.extensions.background_runtime.HostedWorker` for idempotent thread start, cooperative close and explicitly selected failure -retry. This is Python thread IO, not another policy or Agent supervisor. Native +retry. Its `repeat` driver owns reconciliation repetition, next-observation +waits and step-error backoff without rebuilding the caller's live scope. This +is Python thread IO, not another policy or Agent supervisor. Native account observation, manager returns/queued-source recovery and Lark private delivery use it through their existing owners. A normal return is terminal; retry is off unless that owner explicitly enables it. A live thread does not @@ -34,7 +36,13 @@ Keep authorization, election locks, receipts, cursors/backoff and ambiguous-writ recovery with the provider. Never use host restart to replay a model request or discard an in-flight journal. The standalone native account wheel retains its own process entrypoint and needs no Core import for account IO; when hosted by -Core, it delegates thread lifetime to this shared host. +Core, it delegates thread lifetime and observation cadence to this shared host. +The same native step/health callbacks drive standalone and hosted observation; +the small standalone wait loop remains because that distribution supports +account IO without installing Core. Manager startup discovery is not replayed +on a step failure, and Lark retains its existing pool, in-flight futures and +fairness/backoff state while the scanner recovers. Shared waits never imply +model/request replay or establish provider health from thread liveness. The experimental [MemT Nano Turn host](../../packages/loopx-memt-turn/README.md) is a separately installed `generic-cli` execution provider. Its native memory diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 94253626e6..7500ba2d69 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -609,14 +609,14 @@ def run(self): recover_source_chat_provenance(*self.args[:3], cancelled=self.stop.is_set) except (OSError, ValueError, KeyError, TypeError, RuntimeError): logging.getLogger(__name__).warning("Source Chat provenance recovery unavailable") - while not self.stop.is_set(): - try: - drain(*self.args, cancelled=self.stop.is_set) - except (OSError, ValueError, KeyError, TypeError, RuntimeError): - logging.getLogger(__name__).warning( - "Manager return receipts unavailable; retrying" - ) - self.stop.wait(3) + self._return_worker.repeat( + self._return_step, interval_seconds=3, + on_error=lambda: logging.getLogger(__name__).warning( + "Manager return receipts unavailable; retrying"), + ) + + def _return_step(self) -> None: + drain(*self.args, cancelled=self.stop.is_set) def run_executions(self) -> None: # Native previews may wait on a subprocess. Keep result publication @@ -625,12 +625,13 @@ def run_executions(self) -> None: if self._execution_pump is None: self._execution_pump = QueuedSourcePump(*self.args[:3]) execution_pump = self._execution_pump - while not self.stop.is_set(): - try: - execution_pump.tick(cancelled=self.stop.is_set) - except (OSError, ValueError, KeyError, TypeError, RuntimeError): - logging.getLogger(__name__).warning("Queued source execution unavailable; retrying") - self.stop.wait(3) + + def step() -> None: + execution_pump.tick(cancelled=self.stop.is_set) + + self._execution_worker.repeat(step, interval_seconds=3, + on_error=lambda: logging.getLogger(__name__).warning( + "Queued source execution unavailable; retrying")) def close(self) -> None: self.stop.set() diff --git a/loopx/extensions/background_runtime.py b/loopx/extensions/background_runtime.py index 29fe638fb2..4872705bdd 100644 --- a/loopx/extensions/background_runtime.py +++ b/loopx/extensions/background_runtime.py @@ -52,6 +52,38 @@ def _run(self) -> None: if self.stop.wait(self.retry_seconds): return + def repeat(self, step: Callable[[], float | None], *, interval_seconds: float, + on_error: Callable[[], None] | None = None, + wait: Callable[[float], object] | None = None) -> None: + """Drive an opted-in reconciliation step without rebuilding its scope. + + The caller keeps its lease, pool, cursors and receipts around this call. + A returned delay selects the next observation, not a request retry. + Health hooks belong in the step/on_error and observe actual work; this + runner never publishes healthy merely because its thread is alive. + """ + if not math.isfinite(interval_seconds) or interval_seconds <= 0: + raise ValueError("Worker interval must be positive and finite") + while not self.stop.is_set(): + failed = False + try: + requested = step() + delay = interval_seconds if requested is None else requested + if not math.isfinite(delay) or delay <= 0: + raise ValueError("Worker step delay must be positive and finite") + except Exception: + if on_error is not None: + on_error() + else: + logging.getLogger(__name__).warning("Background step failed; provider details withheld") + if self.retry_seconds is None: + return + delay = self.retry_seconds + failed = True + if not self.stop.is_set(): + # A hint may advance a normal scan, never bypass error backoff. + (self.stop.wait if failed else (wait or self.stop.wait))(delay) + def close(self, *, timeout: float = 2) -> None: self.stop.set() thread = self.thread diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index aa83896ef4..312b29aa1a 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -60,7 +60,8 @@ def __init__( self._delivery_wakeup = threading.Event() self._startup_thread: threading.Thread | None = None self._delivery_thread: threading.Thread | None = None - self._delivery_worker: HostedWorker | None = None + self._delivery_worker = HostedWorker(target=self._reconcile_private, stop=self._closed, + name="loopx-lark-private-delivery", retry_seconds=5) def start(self) -> None: """Discover existing bindings without blocking the HTTP readiness path.""" @@ -75,8 +76,6 @@ def start(self) -> None: ) self._startup_thread.start() if self.private_conversations is not None: - self._delivery_worker = HostedWorker(target=self._reconcile_private, stop=self._closed, - name="loopx-lark-private-delivery", retry_seconds=5) self._delivery_worker.start() self._delivery_thread = self._delivery_worker.thread @@ -100,7 +99,8 @@ def _reconcile_private(self) -> None: # existing scanner; it is neither a queue nor a delivery receipt. with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool, \ (wakeup_scope(self._delivery_wakeup) if callable(wakeup_scope) else nullcontext()): - while not self._closed.is_set(): + def step() -> None: + nonlocal attempted, attention, attention_due, reports, reports_due, ingress, ingress_due # Clear BEFORE scanning: an arrival during the scan must also # wake the following wait, rather than disappear in that gap. self._delivery_wakeup.clear() @@ -174,7 +174,10 @@ def _reconcile_private(self) -> None: ingress = pool.submit(observer.observe, cancelled=self._closed.is_set) except Exception: logging.getLogger(__name__).warning("Private Chat delivery discovery is pending") - self._delivery_wakeup.wait(1) + + self._delivery_worker.repeat(step, interval_seconds=1, wait=self._delivery_wakeup.wait, + on_error=lambda: logging.getLogger(__name__).warning( + "Private Chat delivery discovery is pending")) def _admit_private(self, profile: str, event: dict[str, Any]) -> dict[str, Any]: if self._closed.is_set(): @@ -563,8 +566,7 @@ def close(self) -> None: stop.set() for _stop, thread, _fingerprint in workers: thread.join(timeout=3) - if self._delivery_worker is not None: - self._delivery_worker.close(timeout=3) + self._delivery_worker.close(timeout=3) __all__ = ["LarkGoalTopicRuntimeService"] diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py index cb32b3ca2e..a5ff68321d 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py @@ -537,22 +537,31 @@ def run(self): self.stop.wait(60) def _observe(self): + repeat = getattr(self._hosted, "repeat", None) + if callable(repeat): + repeat(self._observe_once, interval_seconds=60, on_error=self._observe_failed) + return + # The standalone wheel has no Core dependency. Its process entrypoint + # drives the same observation/health callbacks as the injected host. while not self.stop.is_set(): - delay = 60.0 try: - self._heartbeat("checking", deadline=480) - value = self.provider.tick() - due = value.get("retry_at") - delay = max(5, min(300, due - time.time())) if due else 60 - self._heartbeat("waiting", deadline=delay + 15, next_check=time.time() + delay) + delay = self._observe_once() except Exception: - # A provider/programming failure must not silently kill the - # thread while the durable policy still promises observation. - # Keep exceptions private and retry under the same owner lock. - self._heartbeat("retrying", deadline=75, next_check=time.time() + 60) + self._observe_failed() delay = 60 self.stop.wait(delay) + def _observe_once(self): + self._heartbeat("checking", deadline=480) + value = self.provider.tick() + due = value.get("retry_at") + delay = max(5, min(300, due - time.time())) if due else 60 + self._heartbeat("waiting", deadline=delay + 15, next_check=time.time() + delay) + return delay + + def _observe_failed(self): + self._heartbeat("retrying", deadline=75, next_check=time.time() + 60) + def _heartbeat(self, phase: str, *, deadline: float, next_check: float | None = None): now = time.time() atomic(self.provider.root / "observer.json", {"schema_version": OBSERVER_SCHEMA, diff --git a/tests/extensions/test_background_runtime.py b/tests/extensions/test_background_runtime.py index 96a50fc6bd..1a668070f6 100644 --- a/tests/extensions/test_background_runtime.py +++ b/tests/extensions/test_background_runtime.py @@ -75,3 +75,80 @@ def test_completed_work_is_not_replayed_by_another_start(): worker.start() assert calls == [True] worker.close() + + +def test_periodic_steps_share_wait_retry_and_keep_their_live_scope(caplog, monkeypatch): + stop = threading.Event() + calls, waits, failures = [], [], [] + scope = object() + + def step(): + calls.append(scope) + if len(calls) == 2: + raise RuntimeError("private synthetic detail") + return 2 if len(calls) == 1 else None + + def wait(delay): + waits.append(delay) + if len(waits) == 3: + stop.set() + + monkeypatch.setattr(stop, "wait", wait) + worker = HostedWorker(name="synthetic-periodic", stop=stop, retry_seconds=.25, + target=lambda: worker.repeat(step, interval_seconds=4, wait=wait, + on_error=lambda: failures.append(True))) + worker.start() + worker.thread.join(2) + assert not worker.thread.is_alive() + assert calls == [scope, scope, scope] and waits == [2, .25, 4] + assert failures == [True] and "private synthetic detail" not in caplog.text + + +def test_periodic_step_failure_is_terminal_without_explicit_retry(): + calls = [] + + def step(): + calls.append(True) + raise RuntimeError("synthetic stop") + + worker = HostedWorker(name="synthetic-periodic-once", stop=threading.Event(), + target=lambda: worker.repeat(step, interval_seconds=1)) + worker.start() + worker.thread.join(2) + worker.start() + assert not worker.thread.is_alive() and calls == [True] + + +def test_invalid_step_delay_uses_failure_backoff_instead_of_spinning(monkeypatch): + stop, waits, failures = threading.Event(), [], [] + + def wait(delay): + waits.append(delay) + stop.set() + + monkeypatch.setattr(stop, "wait", wait) + worker = HostedWorker(name="synthetic-invalid-delay", stop=stop, retry_seconds=3, + target=lambda: worker.repeat(lambda: float("nan"), interval_seconds=1, + wait=wait, on_error=lambda: failures.append(True))) + worker.start() + worker.thread.join(2) + assert waits == [3] and failures == [True] + + +def test_wakeup_hint_cannot_bypass_failure_backoff(monkeypatch): + stop, waits, hints = threading.Event(), [], [] + + def step(): + raise OSError("synthetic scanner failure") + + def backoff(delay): + waits.append(delay) + stop.set() + + monkeypatch.setattr(stop, "wait", backoff) + worker = HostedWorker(name="synthetic-hint", stop=stop, retry_seconds=3, + target=lambda: worker.repeat(step, interval_seconds=1, + wait=lambda delay: hints.append(delay))) + worker.start() + worker.thread.join(2) + assert waits == [3] and hints == [] diff --git a/tests/presentation/test_native_observer_health.py b/tests/presentation/test_native_observer_health.py index 6488fcd27f..14766b0545 100644 --- a/tests/presentation/test_native_observer_health.py +++ b/tests/presentation/test_native_observer_health.py @@ -98,10 +98,11 @@ def test_health_requires_a_live_lock_and_does_not_reuse_a_stale_success(service) assert "private synthetic marker detail" not in json.dumps(provider.observer_health(True)) -def test_unexpected_tick_failure_retries_and_progresses_without_leaking_details(service, monkeypatch): +@pytest.mark.parametrize("hosted", [False, True], ids=["standalone", "shared-core-loop"]) +def test_unexpected_tick_failure_retries_and_progresses_without_leaking_details(service, monkeypatch, hosted): owner, provider, native = service owner.watcher.close() - watcher = native.Watcher(provider) + watcher = owner._new_observer(native) if hosted else native.Watcher(provider) owner.watcher = watcher original_tick, original_atomic, original_wait = provider.tick, native.atomic, watcher.stop.wait phases, calls = [], [] diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index a675f37e82..928448c658 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -3031,6 +3031,57 @@ def reconcile_request(self, path): +def test_private_scanner_recovers_inside_its_live_pool_without_restarting_requests(tmp_path, monkeypatch, caplog): + from loopx.extensions.lark import goal_topic_runtime_service as module + + release, full, recovered = threading.Event(), threading.Event(), threading.Event() + calls, pools, scans = [], [], [] + pool_factory = module.ThreadPoolExecutor + + def make_pool(*args, **kwargs): + pool = pool_factory(*args, **kwargs) + pools.append(pool) + return pool + + class Pending: + def pending_delivery_paths(self): + return [tmp_path / str(index) for index in range(9)] + + def reconcile_request(self, path): + calls.append(path) + if len(calls) == 4: + full.set() + assert release.wait(10) + return 0 + + monkeypatch.setattr(module, "ThreadPoolExecutor", make_pool) + service = module.LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path, + runtime_controller=None, private_conversations=Pending()) + clear = service._delivery_wakeup.clear + + def fail_once(): + scans.append(True) + if len(scans) == 2: + raise AssertionError("private synthetic scanner detail") + clear() + if len(scans) > 2: + recovered.set() + + monkeypatch.setattr(service._delivery_wakeup, "clear", fail_once) + service.start() + service._delivery_worker.retry_seconds = .01 # Only this disposable scanner's failure wait. + try: + assert full.wait(3) + assert recovered.wait(4), "Scanner recovery must not wait for unrelated in-flight writes to drain" + assert len(pools) == 1 and len(calls) == 4 and len(set(calls)) == 4 + assert "private synthetic scanner detail" not in caplog.text + finally: + release.set() + service.close() + service._delivery_thread.join(5) + assert not service._delivery_thread.is_alive() and len(calls) == 4 + + def test_completed_command_recovery_defers_app_probe_and_revocation_blocks_reply(ordinary): # noqa: F811 store, runtime, provider, transport = connect(ordinary) try: diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 7116d617b5..29f756d0b3 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -1266,6 +1266,12 @@ def test_host_recovery_returns_the_original_result_once(flow, monkeypatch): report(root, "research", "worker", rid, "conclusion", "Original result after host recovery.") original, attempts, sends = roundtrip.drain, [], [] sent = threading.Event() + provenance = roundtrip.recover_source_chat_provenance + starts = [] + + def discover(*args, **kwargs): + starts.append(True) + return provenance(*args, **kwargs) def failed_once(*args, **kwargs): attempts.append(True) @@ -1279,6 +1285,7 @@ def transport(*_): return {"reply_verified": True, "idempotency_key": "sha256:synthetic-proof"} monkeypatch.setattr(roundtrip, "drain", failed_once) + monkeypatch.setattr(roundtrip, "recover_source_chat_provenance", discover) service = ReturnService(root, registry, store, transport) service.start() service.start() @@ -1288,6 +1295,7 @@ def transport(*_): service.close() drain(root, registry, store, transport) assert sends == [True] + assert starts == [True], "Step recovery must retain its scope instead of replaying startup discovery" state = json.loads((_root(root) / "replies" / rid / "conclusion.delivery.json").read_text()) assert state["status"] == "delivered" From 77c95fe432d1aaa592d1604be36bab2cc52a455f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 17:17:07 +0530 Subject: [PATCH 2/8] docs(runtime): distinguish host completion from repeated steps Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/extensions.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/reference/extensions.md b/docs/reference/extensions.md index 8c07eb62b4..11ef677c48 100644 --- a/docs/reference/extensions.md +++ b/docs/reference/extensions.md @@ -29,8 +29,9 @@ retry. Its `repeat` driver owns reconciliation repetition, next-observation waits and step-error backoff without rebuilding the caller's live scope. This is Python thread IO, not another policy or Agent supervisor. Native account observation, manager returns/queued-source recovery and Lark private -delivery use it through their existing owners. A normal return is terminal; -retry is off unless that owner explicitly enables it. A live thread does not +delivery use it through their existing owners. A host target's normal return is +terminal; an explicitly repeated step returning `None` uses its declared interval. +Failure retry is off unless that owner explicitly enables it. A live thread does not establish provider health, and waiting, cooldown or disabled work is not a fault. Keep authorization, election locks, receipts, cursors/backoff and ambiguous-write recovery with the provider. Never use host restart to replay a model request or From 814f0e65613665af64fe5d30b13307a13a9d1b47 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:15:58 +0530 Subject: [PATCH 3/8] feat(runtime): supervise registered background services and project progress health Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/extensions.md | 22 +++ .../capabilities/manager_context/roundtrip.py | 5 +- loopx/chat_server.py | 22 ++- .../control_plane/effect_runtime_handlers.ts | 1 + .../runtime/background_services.ts | 93 +++++++++++ loopx/extensions/background_runtime.py | 55 ++++++- loopx/extensions/background_services.py | 117 ++++++++++++++ .../lark/goal_topic_runtime_service.py | 15 +- loopx/presentation/background_services_api.py | 40 +++++ loopx/presentation/codex_accounts_api.py | 27 +++- .../background_services.test.ts | 43 +++++ tests/extensions/test_background_services.py | 115 +++++++++++++ .../test_background_services_api.py | 153 ++++++++++++++++++ tsconfig.control-plane.json | 1 + 14 files changed, 687 insertions(+), 22 deletions(-) create mode 100644 loopx/control_plane/runtime/background_services.ts create mode 100644 loopx/extensions/background_services.py create mode 100644 loopx/presentation/background_services_api.py create mode 100644 tests/control_plane_ts/background_services.test.ts create mode 100644 tests/extensions/test_background_services.py create mode 100644 tests/presentation/test_background_services_api.py diff --git a/docs/reference/extensions.md b/docs/reference/extensions.md index 11ef677c48..36cd4a4608 100644 --- a/docs/reference/extensions.md +++ b/docs/reference/extensions.md @@ -45,6 +45,28 @@ on a step failure, and Lark retains its existing pool, in-flight futures and fairness/backoff state while the scanner recovers. Shared waits never imply model/request replay or establish provider health from thread liveness. +The Chat process registers these four loops with the process-scoped +`BackgroundServices` lifecycle owner. It reconciles existing owner intent at +startup and every 15 seconds, isolates owner failures, and fences shutdown +before closing registered services. There is no additional saved enable switch, +job queue, system daemon or permission grant. The App/OS still supervises the +Chat process. An enabled native observer that stopped is re-created from its +current saved policy; a live scan or a normally completed one-shot is not replayed. + +Settings → **Background services** reads the same content-free projection as +`GET /api/chat/background-services`. **Recheck** uses the loopback-protected +`POST` endpoint with `{"operation":"recheck"}` to reconcile current owner intent +and read back observations. Reads are passive. The TypeScript owner +`control_plane/runtime/background_services.ts` distinguishes checking, completed +scans, backoff, overdue progress, unavailable evidence and disabled work. A +progress deadline diagnoses silence; it never kills a thread or authorizes an +effect retry. Successful scanning does not assert successful delivery of every +individual request. Native observation uses its provider's elected-lock health, +including an owner in another process; a passive Core contender cannot replace +that evidence. Its progress timestamp is not reported as a successful quota +sample. Per-request receipts, profile listeners and ambiguous-write recovery +remain with their existing owners. + The experimental [MemT Nano Turn host](../../packages/loopx-memt-turn/README.md) is a separately installed `generic-cli` execution provider. Its native memory is scoped advisory history, not canonical work state or accepted Reward Memory. diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 7500ba2d69..a5bb1a07a5 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -498,7 +498,7 @@ def _pending_return_paths(host, paths): @effect_runtime_request_scope() -def drain(root, registry, store, external_sender, *, now=None, cancelled=lambda: False): +def drain(root, registry, store, external_sender, *, now=None, cancelled=lambda: False, progress=lambda: None): """Run the typed return transaction with the existing Chat/provider adapters.""" if cancelled(): return 0 @@ -511,6 +511,7 @@ def drain(root, registry, store, external_sender, *, now=None, cancelled=lambda: host = {"root": root, "registry": registry, "store": store, "now": now, "exact": exact, "cancelled": cancelled} paths = iter_result_paths(_root(root) / "replies") for path in _pending_return_paths(host, paths): + progress() if cancelled(): break if not _belongs_to_chat_host(root, store, path): @@ -616,7 +617,7 @@ def run(self): ) def _return_step(self) -> None: - drain(*self.args, cancelled=self.stop.is_set) + drain(*self.args, cancelled=self.stop.is_set, progress=self._return_worker.checkpoint) def run_executions(self) -> None: # Native previews may wait on a subprocess. Keep result publication diff --git a/loopx/chat_server.py b/loopx/chat_server.py index a737c61852..24c79f06f9 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -95,6 +95,10 @@ ) from .extensions.lark.goal_topic_connections import list_lark_apps from .extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService +from .extensions.background_services import BackgroundServices +from .presentation.background_services_api import ( + BACKGROUND_SERVICES_PATH, BackgroundServicesRequestMixin, register_chat_background_services, +) from .extensions.lark.cli_transport import CommandRunner from .extensions.runtime import ( default_extension_state_file, @@ -478,19 +482,22 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self.runtime_identity = release_runtime_identity() super().__init__(*args, **kwargs) self.completed_todo_pages = CompletedTodoPages() + self.background_services = BackgroundServices() + self.background_service_owners: set[str] = set() def server_close(self) -> None: - if hasattr(self, "codex_account_service"): + self.background_services.close() + if hasattr(self, "codex_account_service") and "codex_account_service" not in self.background_service_owners: self.codex_account_service.close() if hasattr(self, "lark_app_setup_manager"): self.lark_app_setup_manager.close() - if hasattr(self, "manager_return_service"): + if hasattr(self, "manager_return_service") and "manager_return_service" not in self.background_service_owners: self.manager_return_service.close() if hasattr(self, "conversation_transports"): self.conversation_transports.close() if hasattr(self, "delegation_wake_service"): self.delegation_wake_service.close() - if hasattr(self, "lark_goal_topic_runtime"): + if hasattr(self, "lark_goal_topic_runtime") and "lark_goal_topic_runtime" not in self.background_service_owners: self.lark_goal_topic_runtime.close() if hasattr(self, "runtime_controller"): self.runtime_controller.close() @@ -498,6 +505,7 @@ def server_close(self) -> None: class ChatRequestHandler( + BackgroundServicesRequestMixin, ChatModelSettingsRequestMixin, GoalSourceRequestMixin, GoalAcceptanceRequestMixin, @@ -1699,6 +1707,7 @@ def do_GET(self) -> None: } ) get_dispatch: dict[str, Callable[[], None]] = { + BACKGROUND_SERVICES_PATH: self._background_services_read, "/api/chat/request-inventory": self._request_inventory_read, GOAL_SOURCE_LINKS_PATH: self._goal_source_links, "/api/chat/todo/detail": self._todo_detail, @@ -1763,6 +1772,7 @@ def do_POST(self) -> None: if self._dispatch_zcode_goal(path, apply=True): return post_dispatch: dict[str, Callable[[], None]] = { + BACKGROUND_SERVICES_PATH: self._background_services_recheck, GOAL_SOURCE_LINKS_PATH: self._goal_source_links, RESEARCH_ANALYSIS_PATH: self._research_analysis_write, RESEARCH_ANALYSIS_PATH + "/cancel": self._research_analysis_write, @@ -1904,7 +1914,7 @@ def serve_chat( resolve_runtime_root(registry, registry_path=resolved_registry_path) if registry.get("common_runtime_root") else runtime_root ) - server.codex_account_service = CodexAccountService(account_runtime_root) + server.codex_account_service = CodexAccountService(account_runtime_root, start_observer=False) server.runtime_root_override = resolved_runtime_root_override server.scan_roots = resolved_scan_roots server.limit = limit @@ -1997,7 +2007,6 @@ def _lark_snapshot() -> dict[str, object]: runtime_controller=server.runtime_controller, ), ) - server.lark_goal_topic_runtime.start() from .extensions.lark.manager_returns import start_return_service server.manager_return_service = start_return_service( server, server.runtime_controller.coordination_runtime_root, start_service=False @@ -2027,7 +2036,8 @@ def _wake_goal_context(session: dict[str, object]) -> dict[str, object]: try: if external_conversation_factories: server.conversation_transports.start(server) - server.manager_return_service.start() + register_chat_background_services(server) + server.background_services.start() server.serve_forever() except KeyboardInterrupt: print("Stopping LoopX Chat", flush=True) diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 61f875d2df..125e4caeef 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -695,6 +695,7 @@ export function createEffectRuntimeHandlers( ["collaboration.result.attachment_refs", lazyHandler(() => import("./collaboration/result_publication.ts"), ({resultAttachmentRefs}) => resultAttachmentRefs)], ["collaboration.result.delivery_ready", lazyHandler(() => import("./collaboration/result_publication.ts"), ({collaborationResultDeliveryReady}) => collaborationResultDeliveryReady)], ["collaboration.peer_host_route.select", lazyHandler(() => import("./collaboration/peer_route_selection.ts"), ({selectObservedPeerHostRoute}) => selectObservedPeerHostRoute)], + ["runtime.background.project", lazyHandler(() => import("./runtime/background_services.ts"), ({projectBackgroundServices}) => projectBackgroundServices)], ["runtime.execution_identity.codex", lazyHandler(() => import("./runtime/execution_identity.ts"), ({readCodexExecutionIdentity}) => readCodexExecutionIdentity)], ["runtime.execution_identity.match", lazyHandler(() => import("./runtime/execution_identity.ts"), ({matchExecutionDeclaration}) => matchExecutionDeclaration)], [ diff --git a/loopx/control_plane/runtime/background_services.ts b/loopx/control_plane/runtime/background_services.ts new file mode 100644 index 0000000000..8588971bbb --- /dev/null +++ b/loopx/control_plane/runtime/background_services.ts @@ -0,0 +1,93 @@ +/** Process-local background diagnostics. No scheduling or effect admission. + * This is the shared projection contract; provider-specific health remains an + * optional typed probe. Thread liveness is never evidence of a completed scan. + */ +import {requireJsonObject, requireBoolean, requireInteger, requireNonEmptyString} from "../runtime_decode.ts"; + +export const BACKGROUND_STATES = ["disabled", "unavailable", "unverified", "starting", + "checking", "healthy", "retrying", "overdue", "stopping", "stopped", "completed"] as const; +export type BackgroundState = typeof BACKGROUND_STATES[number]; + +type Worker = { + alive: boolean; stop_requested: boolean; terminal: boolean; executing: boolean; + started_at: number | null; last_progress_at?: number | null; last_success_at: number | null; last_failure_at: number | null; + next_check_at: number | null; deadline_at: number | null; failure_count: number; +}; +type Observation = { + enabled: boolean | null; available: boolean; reconcile_failed?: boolean; + worker?: Worker; + probe?: {state: BackgroundState; checked_at: number | null; next_check_at: number | null}; +}; +const epoch = (value: unknown): number | null => + typeof value === "number" && Number.isFinite(value) && value > 0 ? value : null; + +export function projectBackgroundTask(input: Observation, now: number) { + const worker = input.worker; + let state: BackgroundState; + if (!input.available) state = "unavailable"; + else if (input.enabled === false) state = "disabled"; + else if (input.enabled !== true) state = "unverified"; + else if (input.probe) state = BACKGROUND_STATES.includes(input.probe.state) + ? input.probe.state : "unverified"; + else if (!worker) state = "unverified"; + else if (worker.stop_requested) state = worker.alive ? "stopping" : "stopped"; + else if ([worker.started_at, worker.last_progress_at, worker.last_success_at, worker.last_failure_at].some(value => (epoch(value) ?? 0) > now + 5)) state = "unverified"; + else if (!worker.alive) state = worker.terminal && worker.failure_count === 0 + && epoch(worker.last_success_at) !== null ? "completed" : "stopped"; + else if (epoch(worker.started_at) === null) state = "starting"; + else if (epoch(worker.deadline_at) === null) state = "unverified"; + else if (worker.deadline_at! < now) state = "overdue"; + else if (worker.failure_count > 0) state = "retrying"; + else if (worker.executing) state = "checking"; + else state = epoch(worker.last_success_at) !== null ? "healthy" : "unverified"; + if (input.probe && ["healthy", "checking", "retrying"].includes(state) + && (epoch(input.probe.checked_at) === null || input.probe.checked_at! > now + 5)) state = "unverified"; + // Reconcile failure cannot turn a verified stopped/disabled service green. + // Preserve a more precise provider or progress fault when one is observed. + if (input.reconcile_failed && ["healthy", "checking", "starting"].includes(state)) state = "retrying"; + return {state, observed_at: epoch(input.probe?.checked_at) + ?? (Math.max(epoch(worker?.started_at) ?? 0, epoch(worker?.last_progress_at) ?? 0, epoch(worker?.last_success_at) ?? 0, + epoch(worker?.last_failure_at) ?? 0) || null), + last_success_at: input.probe ? null : epoch(worker?.last_success_at), + last_failure_at: input.probe ? null : epoch(worker?.last_failure_at), + next_check_at: epoch(input.probe?.next_check_at ?? worker?.next_check_at), + failure_count: input.probe ? null : worker?.failure_count ?? 0, + reconcile_failed: input.reconcile_failed === true}; +} + +function observation(value: unknown): Observation { + const row = requireJsonObject(value, "background observation"); + let worker: Worker | undefined; + if (row.worker !== undefined) { + const data = requireJsonObject(row.worker, "worker progress"); + const failures = requireInteger(data.failure_count, "failure count"); + if (failures < 0) throw new Error("Invalid failure count"); + worker = {alive: requireBoolean(data.alive, "alive"), + stop_requested: requireBoolean(data.stop_requested, "stop requested"), + terminal: requireBoolean(data.terminal, "terminal"), executing: requireBoolean(data.executing, "executing"), + started_at: epoch(data.started_at), last_progress_at: epoch(data.last_progress_at), last_success_at: epoch(data.last_success_at), + last_failure_at: epoch(data.last_failure_at), next_check_at: epoch(data.next_check_at), + deadline_at: epoch(data.deadline_at), failure_count: failures}; + } + const probe = row.probe === undefined ? undefined : requireJsonObject(row.probe, "provider probe"); + return {available: requireBoolean(row.available, "available"), + enabled: row.enabled === null ? null : requireBoolean(row.enabled, "enabled"), worker, + reconcile_failed: row.reconcile_failed === undefined ? false : requireBoolean(row.reconcile_failed, "reconcile failed"), + ...(probe ? {probe: {state: probe.state as BackgroundState, checked_at: epoch(probe.checked_at), + next_check_at: epoch(probe.next_check_at)}} : {})}; +} + +export function projectBackgroundServices(params: Record) { + const now = epoch(params.now); + if (now === null || !Array.isArray(params.tasks)) throw new Error("Invalid background observation"); + return {ok: true as const, schema_version: "background_services_v1" as const, checked_at: now, + supervisor: projectBackgroundTask(observation(params.supervisor), now), + tasks: params.tasks.map(value => { + const task = requireJsonObject(value, "background task"); + return {id: requireNonEmptyString(task.id, "task id"), + service_id: requireNonEmptyString(task.service_id, "service id"), + ...projectBackgroundTask(observation(task), now)}; + })}; +} + +export type BackgroundServicesSnapshot = ReturnType; diff --git a/loopx/extensions/background_runtime.py b/loopx/extensions/background_runtime.py index 4872705bdd..261df31686 100644 --- a/loopx/extensions/background_runtime.py +++ b/loopx/extensions/background_runtime.py @@ -9,19 +9,59 @@ import logging import math import threading +import time class HostedWorker: def __init__(self, *, name: str, target: Callable[[], None], - stop: threading.Event, retry_seconds: float | None = None): + stop: threading.Event, retry_seconds: float | None = None, + progress_deadline_seconds: float = 300): if retry_seconds is not None and (not math.isfinite(retry_seconds) or retry_seconds <= 0): raise ValueError("Worker retry delay must be positive and finite") self.name, self.target = name, target self.stop, self.retry_seconds = stop, retry_seconds + if not math.isfinite(progress_deadline_seconds) or progress_deadline_seconds <= 0: + raise ValueError("Worker progress deadline must be positive and finite") + # A diagnostic deadline, never cancellation or permission to replay. + self.progress_deadline_seconds = progress_deadline_seconds self._lock = threading.Lock() self._terminal = False + self._repeating = False + self._progress = {"started_at": None, "last_success_at": None, + "last_failure_at": None, "next_check_at": None, + "last_progress_at": None, "deadline_at": None, + "failure_count": 0, "executing": False} self.thread = self._new_thread() + def snapshot(self) -> dict: + """Passive facts; reading never advances progress or starts work.""" + with self._lock: + return {**self._progress, "alive": self.thread.is_alive(), + "stop_requested": self.stop.is_set(), "terminal": self._terminal} + + def checkpoint(self) -> None: + """Report a completed unit inside a longer scan without declaring success.""" + with self._lock: + if self._progress["executing"]: + self._progress["last_progress_at"] = time.time() + self._progress["deadline_at"] = time.time() + self.progress_deadline_seconds + + def _begin(self) -> None: + with self._lock: + self._progress.update(started_at=time.time(), last_progress_at=time.time(), executing=True, next_check_at=None, + deadline_at=time.time() + self.progress_deadline_seconds) + + def _settle(self, *, failed: bool, delay: float | None) -> None: + now = time.time() + with self._lock: + self._progress.update(executing=False, last_progress_at=now, + next_check_at=now + delay if delay is not None else None, + deadline_at=now + delay + 15 if delay is not None else None) + if failed: + self._progress.update(last_failure_at=now, failure_count=self._progress["failure_count"] + 1) + else: + self._progress.update(last_success_at=now, failure_count=0) + def _new_thread(self) -> threading.Thread: return threading.Thread(target=self._run, daemon=True, name=self.name) @@ -37,13 +77,17 @@ def start(self): def _run(self) -> None: while not self.stop.is_set(): try: + self._begin() self.target() # Normal completion, including a one-shot discovery, is not a # failure and must not be automatically replayed. with self._lock: self._terminal = True + if not self._repeating: + self._settle(failed=False, delay=None) return except Exception: + self._settle(failed=True, delay=self.retry_seconds) logging.getLogger(__name__).warning("Background worker failed; provider details withheld") if self.retry_seconds is None: with self._lock: @@ -59,19 +103,22 @@ def repeat(self, step: Callable[[], float | None], *, interval_seconds: float, The caller keeps its lease, pool, cursors and receipts around this call. A returned delay selects the next observation, not a request retry. - Health hooks belong in the step/on_error and observe actual work; this - runner never publishes healthy merely because its thread is alive. + Progress records actual step completion. A live thread alone never + establishes a successful check or provider/request health. """ if not math.isfinite(interval_seconds) or interval_seconds <= 0: raise ValueError("Worker interval must be positive and finite") + self._repeating = True while not self.stop.is_set(): failed = False try: + self._begin() requested = step() delay = interval_seconds if requested is None else requested if not math.isfinite(delay) or delay <= 0: raise ValueError("Worker step delay must be positive and finite") except Exception: + self._settle(failed=True, delay=self.retry_seconds) if on_error is not None: on_error() else: @@ -80,6 +127,8 @@ def repeat(self, step: Callable[[], float | None], *, interval_seconds: float, return delay = self.retry_seconds failed = True + if not failed: + self._settle(failed=False, delay=delay) if not self.stop.is_set(): # A hint may advance a normal scan, never bypass error backoff. (self.stop.wait if failed else (wait or self.stop.wait))(delay) diff --git a/loopx/extensions/background_services.py b/loopx/extensions/background_services.py new file mode 100644 index 0000000000..083044871a --- /dev/null +++ b/loopx/extensions/background_services.py @@ -0,0 +1,117 @@ +"""Process-scoped lifecycle IO for opted-in background service owners. + +No second persisted configuration, job queue or effect authority lives here. +Owners re-read their grants on reconciliation; TypeScript projects health from +passive observations. The process supervisor remains the App/OS service host. +""" +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +import threading +import time + +from .background_runtime import HostedWorker +from ..control_plane.effect_runtime import effect_runtime_result + + +@dataclass(frozen=True) +class BackgroundTask: + task_id: str + observe: Callable[[], dict] + + +@dataclass(frozen=True) +class BackgroundService: + service_id: str + tasks: tuple[BackgroundTask, ...] + reconcile: Callable[[], None] + close: Callable[[], None] + + +def worker_observation(worker: HostedWorker, *, enabled: bool = True) -> dict: + return {"available": True, "enabled": enabled, "worker": worker.snapshot()} + + +class BackgroundServices: + """One registry supervises existing owners; it never retries their effects.""" + + def __init__(self) -> None: + self._services: dict[str, BackgroundService] = {} + self._errors: set[str] = set() + self._lifecycle = threading.RLock() + self._state_lock = threading.Lock() + self._stop = threading.Event() + self._closed = False + self._started = False + self._supervisor = HostedWorker(name="loopx-background-services", stop=self._stop, + target=self._run, retry_seconds=15) + + def register(self, service: BackgroundService) -> None: + with self._lifecycle, self._state_lock: + ids = [task.task_id for item in self._services.values() for task in item.tasks] + incoming = [task.task_id for task in service.tasks] + if (self._started or self._stop.is_set() or service.service_id in self._services or not incoming + or len(set(incoming)) != len(incoming) or set(ids) & set(incoming)): + raise ValueError("Duplicate or closed background registration") + self._services[service.service_id] = service + + def start(self) -> None: + with self._state_lock: + self._started = True + self._supervisor.start() + + def _run(self) -> None: + self._supervisor.repeat(self.reconcile, interval_seconds=15) + + def reconcile(self) -> None: + with self._lifecycle: + for service in self._services.values(): + if self._stop.is_set(): + return + try: + # Each owner performs its own fresh permission/intent read. + # start() is idempotent; live or completed work is not reset. + service.reconcile() + with self._state_lock: + self._errors.discard(service.service_id) + except Exception: + # Isolate owners and withhold private provider diagnostics. + with self._state_lock: + self._errors.add(service.service_id) + self._supervisor.checkpoint() + + def snapshot(self) -> dict: + # Copy the registry, then observe outside lifecycle mutation. A GET must + # neither repair a stopped task nor manufacture a progress heartbeat. + with self._state_lock: + services, errors = tuple(self._services.values()), self._errors.copy() + rows = [] + for service in services: + for task in service.tasks: + try: + observation = task.observe() + except Exception: + observation = {"available": False, "enabled": None} + rows.append({**observation, "id": task.task_id, "service_id": service.service_id, + "reconcile_failed": service.service_id in errors}) + return effect_runtime_result("runtime.background.project", { + "now": time.time(), "tasks": rows, + "supervisor": worker_observation(self._supervisor), + }) + + def close(self) -> None: + # Fence before joining; even a concurrent recheck cannot resurrect work. + self._stop.set() + self._supervisor.close() + with self._lifecycle: + if self._closed: + return + self._closed = True + for service in reversed(tuple(self._services.values())): + try: + service.close() + except Exception: + with self._state_lock: + self._errors.add(service.service_id) + diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index 312b29aa1a..b29702089f 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -67,14 +67,15 @@ def start(self) -> None: """Discover existing bindings without blocking the HTTP readiness path.""" with self._lock: - if self._closed.is_set() or self._startup_thread is not None: + if self._closed.is_set(): return - self._startup_thread = threading.Thread( - target=self._refresh_on_start, - name="loopx-lark-startup", - daemon=True, - ) - self._startup_thread.start() + if self._startup_thread is None: + self._startup_thread = threading.Thread( + target=self._refresh_on_start, + name="loopx-lark-startup", + daemon=True, + ) + self._startup_thread.start() if self.private_conversations is not None: self._delivery_worker.start() self._delivery_thread = self._delivery_worker.thread diff --git a/loopx/presentation/background_services_api.py b/loopx/presentation/background_services_api.py new file mode 100644 index 0000000000..5d70ef9190 --- /dev/null +++ b/loopx/presentation/background_services_api.py @@ -0,0 +1,40 @@ +"""Chat composition and loopback read/recheck for the common background host.""" +from __future__ import annotations + +from ..extensions.background_services import BackgroundService, BackgroundTask, worker_observation + +BACKGROUND_SERVICES_PATH = "/api/chat/background-services" + + +def register_chat_background_services(server) -> None: + accounts = server.codex_account_service + returns = server.manager_return_service + lark = server.lark_goal_topic_runtime + runtime = server.background_services + runtime.register(BackgroundService("codex-accounts", ( + BackgroundTask("codex-observation", accounts.background_observation), + ), accounts.reconcile_background, accounts.close)) + runtime.register(BackgroundService("manager-context", ( + BackgroundTask("manager-returns", lambda: worker_observation(returns._return_worker)), + BackgroundTask("manager-executions", lambda: worker_observation(returns._execution_worker)), + ), returns.start, returns.close)) + runtime.register(BackgroundService("lark", ( + BackgroundTask("lark-delivery", lambda: worker_observation(lark._delivery_worker, + enabled=lark.private_conversations is not None)), + ), lark.start, lark.close)) + server.background_service_owners = {"codex_account_service", "manager_return_service", "lark_goal_topic_runtime"} + + +class BackgroundServicesRequestMixin: + def _background_services_read(self): + self._send_json(self.server.background_services.snapshot()) + + def _background_services_recheck(self): + try: + valid = self._read_json() == {"operation": "recheck"} + except (ValueError, TypeError): + valid = False + if not valid: + return self._send_json({"ok": False, "code": "invalid_request"}, status=400) + self.server.background_services.reconcile() + self._send_json(self.server.background_services.snapshot()) diff --git a/loopx/presentation/codex_accounts_api.py b/loopx/presentation/codex_accounts_api.py index eff991952e..cdc1492898 100644 --- a/loopx/presentation/codex_accounts_api.py +++ b/loopx/presentation/codex_accounts_api.py @@ -16,7 +16,7 @@ class CodexAccountService: - def __init__(self, runtime_root: Path): + def __init__(self, runtime_root: Path, *, start_observer: bool = True): self.provider: Any = None self.watcher: Any = None self.mutex = threading.Lock() @@ -32,7 +32,8 @@ def __init__(self, runtime_root: Path): # Rehydrate an explicitly enabled observer after an App restart. The # absent/default-off path neither creates state nor reads native auth. try: - self._reconcile_observer() + if start_observer: + self._reconcile_observer() except Exception: self.error = "state_unavailable" @@ -105,10 +106,28 @@ def call(self, operation: str, parameters: dict[str, Any]) -> dict[str, Any]: pass return {**cached, "ok": False, "available": True, "code": code, "error_code": code} + def reconcile_background(self) -> None: + with self.mutex: + self._reconcile_observer() + + def background_observation(self) -> dict: + if self.provider is None: + return {"available": False, "enabled": None} + enabled = self.provider.path.exists() and self.provider._load()["policy"]["enabled"] + observe = getattr(self.provider, "observer_health", None) + # The provider observes its elected lock owner, which can be another + # process. The passive Core contender's thread is not that observation. + probe = observe(enabled) if callable(observe) else {"state": "unverified"} + return {"available": True, "enabled": enabled, "probe": { + "state": "unverified" if probe["state"] == "upgrade_required" else probe["state"], + "checked_at": probe.get("checked_at"), "next_check_at": probe.get("next_check_at"), + }} + def close(self) -> None: self.closed = True - if self.watcher: - self.watcher.close() + with self.mutex: + if self.watcher: + self.watcher.close() class _HTTP(Protocol): diff --git a/tests/control_plane_ts/background_services.test.ts b/tests/control_plane_ts/background_services.test.ts new file mode 100644 index 0000000000..6f2ef139bc --- /dev/null +++ b/tests/control_plane_ts/background_services.test.ts @@ -0,0 +1,43 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {projectBackgroundTask, projectBackgroundServices} from "../../loopx/control_plane/runtime/background_services.ts"; + +const now = 1000; +const worker = {alive: true, stop_requested: false, terminal: false, executing: true, + started_at: 900, last_success_at: null, last_failure_at: null, next_check_at: null, + deadline_at: 1100, failure_count: 0}; +const project = (patch = {}) => projectBackgroundTask({available: true, enabled: true, + worker: {...worker, ...patch}}, now); + +test("live work is checking, successful scan is healthy, stalled live thread is overdue", () => { + assert.equal(project().state, "checking"); + assert.equal(project({executing: false, last_success_at: 950, next_check_at: 1050}).state, "healthy"); + assert.equal(project({deadline_at: 999}).state, "overdue"); + assert.equal(project({deadline_at: null}).state, "unverified"); + assert.equal(project({last_success_at: 2000}).state, "unverified"); +}); + +test("backoff, disabled, terminal completion and shutdown are distinct", () => { + assert.equal(project({executing: false, failure_count: 1, last_failure_at: 980}).state, "retrying"); + assert.equal(project({alive: false, terminal: true, last_success_at: 950}).state, "completed"); + assert.equal(project({alive: false, terminal: true, failure_count: 1}).state, "stopped"); + assert.equal(project({stop_requested: true}).state, "stopping"); + assert.equal(projectBackgroundTask({available: true, enabled: false, worker}, now).state, "disabled"); +}); + +test("provider election probe wins over the passive local contender, including unknown and stale evidence", () => { + const input = {available: true, enabled: true, worker: {...worker, deadline_at: 999}, + probe: {state: "healthy" as const, checked_at: 990, next_check_at: 1050}}; + assert.equal(projectBackgroundTask(input, now).state, "healthy"); + assert.equal(projectBackgroundTask({...input, probe: {...input.probe, state: "overdue"}}, now).state, "overdue"); + assert.equal(projectBackgroundTask({...input, probe: {...input.probe, checked_at: null}}, now).state, "unverified"); +}); + +test("RPC decoder rejects malformed facts and emits only content-free fields", () => { + const task = {id: "synthetic", service_id: "fixture", available: true, enabled: true, worker, + private_content: "must not escape"}; + const input = {now, supervisor: task, tasks: [task]}; + assert.equal(JSON.stringify(projectBackgroundServices(input)).includes("must not escape"), false); + assert.throws(() => projectBackgroundServices({...input, tasks: [{...task, enabled: "true"}]})); + assert.throws(() => projectBackgroundServices({...input, tasks: [{...task, worker: {...worker, alive: "true"}}]})); +}); diff --git a/tests/extensions/test_background_services.py b/tests/extensions/test_background_services.py new file mode 100644 index 0000000000..cd087fab4d --- /dev/null +++ b/tests/extensions/test_background_services.py @@ -0,0 +1,115 @@ +"""A common lifecycle must preserve owner authority and diagnose real progress.""" +import threading +import time + +from loopx.extensions.background_runtime import HostedWorker +from loopx.extensions.background_services import ( + BackgroundService, BackgroundServices, BackgroundTask, worker_observation, +) + + +def test_passive_reads_report_stalled_live_work_without_replay_or_fake_success(monkeypatch): + now = [1000.0] + monkeypatch.setattr(time, "time", lambda: now[0]) + entered, release = threading.Event(), threading.Event() + calls = [] + + def blocked(): + calls.append(True) + entered.set() + release.wait(3) + + worker = HostedWorker(name="synthetic-blocked", target=blocked, stop=threading.Event(), + progress_deadline_seconds=20) + runtime = BackgroundServices() + runtime.register(BackgroundService("fixture", ( + BackgroundTask("blocked", lambda: worker_observation(worker)), + ), worker.start, worker.close)) + try: + runtime.reconcile() + assert entered.wait(1) + assert runtime.snapshot()["tasks"][0]["state"] == "checking" + before = worker.snapshot() + now[0] += 21 + assert runtime.snapshot()["tasks"][0]["state"] == "overdue" + runtime.reconcile() + assert calls == [True] and worker.snapshot() == before + assert runtime.snapshot()["tasks"][0]["last_success_at"] is None + finally: + release.set() + runtime.close() + runtime.reconcile() + assert calls == [True] + + +def test_automatic_reconciliation_isolated_failure_and_shutdown_fence(monkeypatch): + from loopx.extensions import background_services as module + period = HostedWorker.repeat + # Shorten only the common observation cadence, keeping the real thread, + # repetition, error isolation and owner callbacks in this test. + monkeypatch.setattr(module.HostedWorker, "repeat", lambda self, step, **kwargs: + period(self, step, interval_seconds=.01)) + recovered, stop = threading.Event(), threading.Event() + attempts, closed = [], [] + runtime = BackgroundServices() + + def reconcile(): + attempts.append(True) + if len(attempts) == 1: + raise RuntimeError("private diagnostics must not escape") + recovered.set() + + runtime.register(BackgroundService("failed-once", ( + BackgroundTask("task", lambda: {"available": True, "enabled": False}), + ), reconcile, lambda: closed.append(True))) + runtime.register(BackgroundService("other", ( + BackgroundTask("other-task", lambda: {"available": True, "enabled": False}), + ), stop.set, lambda: None)) + runtime.start() + assert recovered.wait(2) and stop.is_set(), "One failing owner must not starve another" + runtime.close() + count = len(attempts) + runtime.close() + runtime.reconcile() + assert len(attempts) == count and closed == [True] + + +def test_owner_rechecks_revocation_before_effect_and_completed_work_stays_terminal(): + grant, calls = [False], [] + worker = HostedWorker(name="synthetic-once", target=lambda: calls.append(True), stop=threading.Event()) + runtime = BackgroundServices() + runtime.register(BackgroundService("fixture", ( + BackgroundTask("task", lambda: worker_observation(worker, enabled=grant[0])), + ), lambda: worker.start() if grant[0] else None, worker.close)) + runtime.reconcile() + assert runtime.snapshot()["tasks"][0]["state"] == "disabled" and calls == [] + grant[0] = True + runtime.reconcile() + worker.thread.join(2) + assert runtime.snapshot()["tasks"][0]["state"] == "completed" + runtime.reconcile() + assert calls == [True] + runtime.close() + + +def test_a_stuck_reconciler_cannot_block_its_own_health_readback(): + entered, release, observed = threading.Event(), threading.Event(), threading.Event() + runtime = BackgroundServices() + + def reconcile(): + entered.set() + release.wait(3) + + runtime.register(BackgroundService("fixture", ( + BackgroundTask("task", lambda: {"available": True, "enabled": False}), + ), reconcile, lambda: None)) + runtime.start() + assert entered.wait(1) + reader = threading.Thread(target=lambda: (runtime.snapshot(), observed.set())) + try: + reader.start() + assert observed.wait(1), "Diagnostics must not share the blocked mutation lock" + finally: + release.set() + reader.join(2) + runtime.close() diff --git a/tests/presentation/test_background_services_api.py b/tests/presentation/test_background_services_api.py new file mode 100644 index 0000000000..8866f091e8 --- /dev/null +++ b/tests/presentation/test_background_services_api.py @@ -0,0 +1,153 @@ +"""Real Chat HTTP composition, three production owners and disposable state.""" +import http.client +import importlib +import json +import os +from pathlib import Path +import threading +import time +from types import SimpleNamespace + +import pytest + +from loopx.chat_server import ChatHTTPServer, ChatRequestHandler +from loopx.chat_store import ChatSessionStore +from loopx.capabilities.manager_context.roundtrip import ReturnService +from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService +from loopx.presentation.background_services_api import register_chat_background_services +from loopx.presentation.codex_accounts_api import CodexAccountService + + +@pytest.fixture +def host(tmp_path, monkeypatch): + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / + "packages/loopx-codex-provider-routing/src")) + native = importlib.import_module("loopx_codex_provider_routing.native") + def transport(_adapter, operation, **parameters): + assert operation == "status", "No native IO is authorized by this fixture" + return {"slots": ["a"], "active_slot": "a", "selection_revision": "synthetic", + "current_login_matches": True, "recovery_required": False} + + provider = native.NativeAccounts(tmp_path / "accounts", transport=transport) + monkeypatch.setattr(native, "NativeAccounts", lambda _root: provider) + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.verbose = False + server.codex_account_service = CodexAccountService(tmp_path, start_observer=False) + server.manager_return_service = ReturnService(tmp_path / "runtime", tmp_path / "registry.json", + ChatSessionStore(tmp_path / "runtime"), lambda *_: pytest.fail("No send is authorized")) + server.lark_goal_topic_runtime = LarkGoalTopicRuntimeService( + snapshot_provider=lambda: {"apps": [], "connections": [], "private_profiles": []}, + runtime_root=tmp_path / "runtime", runtime_controller=SimpleNamespace(), action_service=None) + register_chat_background_services(server) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server, provider, native + finally: + server.shutdown() + thread.join(2) + server.server_close() + + +def request(server, method="GET", body=None, origin=None): + connection = http.client.HTTPConnection("127.0.0.1", server.server_port, timeout=10) + headers = {"Origin": origin or f"http://127.0.0.1:{server.server_port}", "Content-Type": "application/json"} + connection.request(method, "/api/chat/background-services", body=json.dumps(body) if body else None, headers=headers) + response = connection.getresponse() + result = response.status, json.loads(response.read()) + connection.close() + return result + + +def test_common_read_is_passive_and_default_off_stays_isolated(host): + server, provider, _ = host + for _ in range(2): + status, result = request(server) + assert status == 200 and len(result["tasks"]) == 4 + tasks = {item["id"]: item for item in result["tasks"]} + assert tasks["codex-observation"]["state"] == "disabled" + assert tasks["lark-delivery"]["state"] == "disabled" + assert tasks["manager-returns"]["state"] == "stopped" + assert not provider.root.exists(), "Reading the default-off path must not create native state" + assert not server.manager_return_service.thread.is_alive() + + +def test_http_recheck_starts_existing_owners_once_and_preserves_disabled_features(host): + server, provider, _ = host + status, _ = request(server, "POST", {"operation": "recheck"}) + assert status == 200 + first = server.manager_return_service.thread + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + _, result = request(server) + tasks = {item["id"]: item for item in result["tasks"]} + if tasks["manager-returns"]["last_success_at"] is not None: + break + time.sleep(.02) + assert tasks["manager-returns"]["state"] == "healthy" + assert tasks["manager-executions"]["state"] == "healthy" + request(server, "POST", {"operation": "recheck"}) + assert server.manager_return_service.thread is first + assert not provider.root.exists() and not server.lark_goal_topic_runtime._delivery_thread + server.background_services.close() + request(server, "POST", {"operation": "recheck"}) + assert not first.is_alive(), "Recheck after shutdown cannot resurrect work" + + +def test_recheck_uses_the_existing_loopback_guard_and_rejects_new_settings(host): + server, _, _ = host + assert request(server, "POST", {"operation": "recheck"}, "https://external.invalid")[0] == 403 + assert request(server, "POST", {"operation": "recheck", "enabled": True})[0] == 400 + assert not server.manager_return_service.thread.is_alive() + + +@pytest.mark.skipif(os.name != "posix", reason="Native elected observation uses POSIX locks") +def test_common_native_health_uses_the_external_elected_owner_not_local_thread(host, monkeypatch): + server, provider, native = host + monkeypatch.setattr(native, "discover", lambda: {"synthetic": True}) + provider.configure(expected_revision=0, enabled=True, slots=["a"]) + # Real standalone lock election and heartbeat; the synthetic step never + # reads native credentials, switches accounts or calls a model. + monkeypatch.setattr(provider, "tick", lambda: {"retry_at": time.time() + 60}) + elected = native.Watcher(provider).start() + try: + deadline = time.monotonic() + 5 + while provider.observer_health(True)["state"] != "healthy" and time.monotonic() < deadline: + time.sleep(.02) + assert server.codex_account_service.watcher.thread is None + before = (provider.root / "observer.json").read_bytes() + _, result = request(server) + task = next(row for row in result["tasks"] if row["id"] == "codex-observation") + assert task["state"] == "healthy" + assert task["last_success_at"] is None, "A progress heartbeat is not an invented successful quota sample" + assert before == (provider.root / "observer.json").read_bytes() + finally: + elected.close() + + +@pytest.mark.skipif(os.name != "posix", reason="Native elected observation uses POSIX locks") +def test_supervisor_restores_enabled_observer_without_a_status_poll_or_policy_edit(host, monkeypatch): + server, provider, native = host + monkeypatch.setattr(native, "discover", lambda: {"synthetic": True}) + monkeypatch.setattr(provider, "tick", lambda: {"retry_at": time.time() + 60}) + provider.configure(expected_revision=0, enabled=True, slots=["a"]) + saved = provider.path.read_bytes() + server.background_services.start() + deadline = time.monotonic() + 5 + while provider.observer_health(True)["state"] != "healthy" and time.monotonic() < deadline: + time.sleep(.02) + assert provider.observer_health(True)["state"] == "healthy" + original = server.codex_account_service.watcher + original.close() + # Use the production 15s supervisor cadence. No status/recheck endpoint + # starts the replacement during this wait. + deadline = time.monotonic() + 20 + while server.codex_account_service.watcher is original and time.monotonic() < deadline: + time.sleep(.05) + assert server.codex_account_service.watcher is not original + deadline = time.monotonic() + 5 + while provider.observer_health(True)["state"] != "healthy" and time.monotonic() < deadline: + time.sleep(.02) + assert provider.observer_health(True)["state"] == "healthy" + assert provider.path.read_bytes() == saved + diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index 72e0c2b7bd..b211b87e2b 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -15,6 +15,7 @@ "include": [ "loopx/control_plane/**/*.ts", "loopx/zcode_goal_mode/**/*.ts", + "tests/control_plane_ts/background_services.test.ts", "tests/control_plane_ts/explore_research.test.ts", "tests/control_plane_ts/explore_research_execution.test.ts", "tests/control_plane_ts/usage_statistics.test.ts", From 26a00fa1ac3dbd90ebc1f9030eeaf9a1d1fe4b57 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:16:26 +0530 Subject: [PATCH 4/8] feat(settings): expose shared background health and recheck recovery Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../dashboard/smoke/settings-target-smoke.mjs | 3 + .../background-services-settings.css | 20 ++++ .../background-services-settings.tsx | 105 ++++++++++++++++++ .../personal-workspace-model.ts | 2 +- .../workspace-settings-page.tsx | 8 +- .../workspace-settings-route.ts | 2 +- 6 files changed, 137 insertions(+), 3 deletions(-) create mode 100644 apps/presentation/dashboard/src/features/personal-workspace/background-services-settings.css create mode 100644 apps/presentation/dashboard/src/features/personal-workspace/background-services-settings.tsx diff --git a/apps/presentation/dashboard/smoke/settings-target-smoke.mjs b/apps/presentation/dashboard/smoke/settings-target-smoke.mjs index fe2c10ec5b..dd2de59a9a 100644 --- a/apps/presentation/dashboard/smoke/settings-target-smoke.mjs +++ b/apps/presentation/dashboard/smoke/settings-target-smoke.mjs @@ -24,6 +24,9 @@ for (const initialGoalId of [undefined, "", "unknown-goal"]) { const machine = render({scope: "machine", initialTab: "language"}); assert.doesNotMatch(machine, /Target Goal|Task ownership|Automation cadence/); assert.match(machine, /This device/); +assert.match(machine, /Background services/); +assert.deepEqual(readSettingsRoute(new URL("http://localhost/chat/?settings_scope=machine&settings_tab=background")), + {kind: "settings", scope: "machine", tab: "background"}); assert.deepEqual(readSettingsRoute(new URL("http://localhost/chat/?settings_scope=goal&settings_goal=shared-goal&settings_tab=ownership")), {kind: "settings", scope: "goal", goalId: "shared-goal", tab: "ownership"}); assert.deepEqual(readSettingsRoute(new URL("http://localhost/chat/?goalId=background&settings_scope=goal")), diff --git a/apps/presentation/dashboard/src/features/personal-workspace/background-services-settings.css b/apps/presentation/dashboard/src/features/personal-workspace/background-services-settings.css new file mode 100644 index 0000000000..7991083c73 --- /dev/null +++ b/apps/presentation/dashboard/src/features/personal-workspace/background-services-settings.css @@ -0,0 +1,20 @@ +.background-settings { display: grid; gap: 16px; max-width: 850px; } +.background-toolbar, .background-actions, .background-task-title { display: flex; align-items: center; justify-content: space-between; gap: 12px; flex-wrap: wrap; } +.background-toolbar > div, .background-status { display: inline-flex; align-items: center; gap: 8px; } +.background-toolbar > .background-actions { flex-wrap: nowrap; flex-shrink: 0; } +.background-actions .personal-secondary-action { display: inline-flex; align-items: center; justify-content: center; gap: 8px; padding: 8px 12px; margin: 0; width: auto; min-height: 44px; } +.background-status { font-size: var(--text-body); white-space: nowrap; color: var(--pw-muted); } +.background-status[data-state="healthy"] { color: var(--pw-green); } +.background-status[data-state="overdue"], .background-status[data-state="retrying"] { color: var(--pw-amber); } +.background-status[data-state="stopped"] { color: var(--pw-red); } +.background-task-list { display: grid; gap: 12px; } +.background-task { border: 1px solid var(--pw-line); border-radius: 12px; padding: 18px 20px; min-width: 0; } +.background-task h2 { font-size: 1rem; font-weight: 600; margin: 0; overflow-wrap: anywhere; } +.background-task-times { display: flex; flex-wrap: wrap; gap: 12px 32px; margin: 16px 0 0; } +.background-task-times div { min-width: 0; } +.background-task-times dt { color: var(--pw-muted); font-size: var(--text-caption); margin-bottom: 4px; } +.background-task-times dd { margin: 0; overflow-wrap: anywhere; } +.background-task details { margin-top: 14px; } +.background-task summary { cursor: pointer; font-size: var(--text-body); color: var(--pw-muted); } +.background-feature-link { border: 0; padding: 12px 0 0; color: var(--pw-blue-ink); background: transparent; font: inherit; cursor: pointer; } +@media (max-width: 560px) { .background-task { padding: 16px; } .background-task-times { gap: 12px 20px; } } diff --git a/apps/presentation/dashboard/src/features/personal-workspace/background-services-settings.tsx b/apps/presentation/dashboard/src/features/personal-workspace/background-services-settings.tsx new file mode 100644 index 0000000000..6ca9e813cf --- /dev/null +++ b/apps/presentation/dashboard/src/features/personal-workspace/background-services-settings.tsx @@ -0,0 +1,105 @@ +import {useCallback, useEffect, useRef, useState} from "react"; +import {AlertCircle, CheckCircle2, Circle, RefreshCw} from "lucide-react"; +import {z} from "zod"; +import {BACKGROUND_STATES} from "../../../../../../loopx/control_plane/runtime/background_services.ts"; +import type {BackgroundState, BackgroundServicesSnapshot} from "../../../../../../loopx/control_plane/runtime/background_services.ts"; +import {requestJson} from "../../data/chat"; +import {ContextHelp} from "./context-help"; +import {useWorkspaceI18n} from "./i18n"; +import "./background-services-settings.css"; + +const path = "/api/chat/background-services"; +const healthSchema = z.object({state: z.enum(BACKGROUND_STATES), + observed_at: z.number().finite().nullable(), last_success_at: z.number().finite().nullable(), + last_failure_at: z.number().finite().nullable(), next_check_at: z.number().finite().nullable(), + failure_count: z.number().int().nonnegative().nullable(), reconcile_failed: z.boolean()}); +const snapshotSchema = z.object({ok: z.literal(true), schema_version: z.literal("background_services_v1"), + checked_at: z.number().finite(), supervisor: healthSchema, + tasks: z.array(healthSchema.extend({id: z.string().min(1), service_id: z.string().min(1)}))}); +const labels: Record = { + healthy: ["正常", "Healthy"], checking: ["正在检查", "Checking"], starting: ["正在启动", "Starting"], + retrying: ["退避重试中", "Backing off"], overdue: ["检查逾期", "Check overdue"], + stopped: ["已停止", "Stopped"], stopping: ["正在停止", "Stopping"], completed: ["已完成", "Completed"], + disabled: ["未启用", "Disabled"], unavailable: ["不可用", "Unavailable"], unverified: ["尚未核实", "Unverified"], +}; +const names: Record = { + "codex-observation": ["Codex 账号与额度", "Codex accounts and quota"], + "manager-returns": ["结果回传", "Result delivery"], + "manager-executions": ["排队执行恢复", "Queued execution recovery"], + "lark-delivery": ["Lark 消息投递", "Lark message delivery"], +}; +const faults: BackgroundState[] = ["overdue", "retrying", "stopped", "unavailable", "unverified"]; + +export function BackgroundServicesSettings({onOpenSettings}: {onOpenSettings?: (tab: "provider" | "lark") => void}) { + const {locale} = useWorkspaceI18n(); + const zh = locale === "zh-CN", language = zh ? 0 : 1; + const [snapshot, setSnapshot] = useState(null); + const [error, setError] = useState(false); + const [busy, setBusy] = useState(false); + const [rechecked, setRechecked] = useState(false); + const generation = useRef(0); + const mounted = useRef(true); + const load = useCallback(async (recheck = false) => { + const current = ++generation.current; + try { + const result = snapshotSchema.parse(await requestJson(path, recheck ? { + method: "POST", body: JSON.stringify({operation: "recheck"}), headers: {"Content-Type": "application/json"}, + } : undefined)); + if (mounted.current && current === generation.current) { + setSnapshot(result); setError(false); if (recheck) setRechecked(true); + } + } catch { + if (mounted.current && current === generation.current) setError(true); + } + }, []); + useEffect(() => { + mounted.current = true; + void load(); + const timer = window.setInterval(() => {if (!document.hidden) void load();}, 10_000); + return () => {mounted.current = false; generation.current++; window.clearInterval(timer);}; + }, [load]); + const time = (value: number | null) => value === null ? "—" : new Date(value * 1000).toLocaleString(locale); + const status = (state: BackgroundState) => { + const Icon = faults.includes(state) ? AlertCircle : state === "healthy" ? CheckCircle2 : Circle; + return ; + }; + return
+
+
{snapshot ? <>{zh ? "后台宿主" : "Background host"} {status(snapshot.supervisor.state)} : null}
+
+ +

{zh ? "重新检查会恢复已启用但未运行的后台任务。授权仍由原设置管理；在途任务不会重放。检查逾期表示近期进度不足，需要查看对应模块。宿主随 LoopX 服务运行，服务进程由 App 和系统管理。" : "Recheck restores enabled background tasks that are not running. Existing settings retain authority; in-flight work is not replayed. Overdue checks need attention in the owning feature. The App and OS manage the LoopX service process."}

+ + +
+
+ {error ?

{zh ? "暂时无法读取后台状态。显示的记录可能已过期，请重新检查。" : "Background status is unavailable. Displayed observations may be stale; recheck to recover."}

: null} + {!snapshot && !error ?

{zh ? "正在读取后台状态…" : "Loading background status…"}

: null} + {rechecked && !error ?

{zh ? "已重新核对启用状态，下面显示最新观测。" : "Enabled state rechecked. Latest observations are shown below."}

: null} + {snapshot?.tasks.length === 0 ?

{zh ? "当前服务未登记后台任务。" : "No background tasks are registered in this service."}

: null} +
+ {snapshot?.tasks.map(task =>
+

{names[task.id]?.[language] ?? task.id}

{status(task.state)}
+
+
{zh ? "最近进度" : "Last progress"}
{time(task.observed_at)}
+
{zh ? "下次检查" : "Next check"}
{time(task.next_check_at)}
+
+
{zh ? "检查记录" : "Check details"} +
+
{zh ? "最近成功" : "Last success"}
{time(task.last_success_at)}
+
{zh ? "最近失败" : "Last failure"}
{time(task.last_failure_at)}
+
{zh ? "连续失败" : "Consecutive failures"}
{task.failure_count ?? "—"}
+
+ {task.reconcile_failed ?

{zh ? "启用状态核对失败，将自动重试。" : "Enabled state reconciliation failed; it will retry automatically."}

: null} + {onOpenSettings && (task.service_id === "codex-accounts" || task.service_id === "lark") ? : null} +
+
)} +
+
; +} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts index 01ab4c1210..5e45241f1a 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts @@ -533,7 +533,7 @@ export type WorkspaceDrawerSelection = | { item: WorkspaceRun; kind: "run" } | { item: WorkspaceOutput; kind: "output" } | { item: WorkspaceActionPreview; kind: "proposal" } - | { goalId?: string; kind: "settings"; scope?: "machine" | "goal"; tab?: "appearance" | "capabilities" | "language" | "lark" | "machine" | "provider" | "steward" | "conversation" | "cadence" | "ownership" } + | { goalId?: string; kind: "settings"; scope?: "machine" | "goal"; tab?: "appearance" | "capabilities" | "language" | "lark" | "machine" | "provider" | "steward" | "conversation" | "cadence" | "ownership" | "background" } | { item: WorkspaceSchedule; kind: "schedule"; diff --git a/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx index 04c7771f22..b8ff4e5cfc 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx @@ -19,6 +19,7 @@ import { MachineConfigurationSettings } from "./machine-configuration-settings"; import { ConfigurationBackupSettings } from "./configuration-backup-settings"; import { OperatorCredentialSettings } from "./operator-credential-settings"; import { CodexAccountSettings } from "./codex-account-settings"; +import { BackgroundServicesSettings } from "./background-services-settings"; import type { PersonalWorkspaceCallbacks, WorkspaceGoal, WorkspaceGoalNotification } from "./personal-workspace-model"; import type { WorkspaceTheme } from "./workspace-theme"; import type { AttentionNotificationSupport } from "./attention-notifications"; @@ -30,11 +31,12 @@ const attentionNotificationSupportCopy = { unsupported: "settings.attentionNotificationsUnsupported", } as const; -type WorkspaceSettingsTab = "ownership" | "steward" | "conversation" | "provider" | "machine" | "capabilities" | "cadence" | "lark" | "appearance" | "language"; +type WorkspaceSettingsTab = "ownership" | "steward" | "conversation" | "provider" | "machine" | "capabilities" | "cadence" | "lark" | "appearance" | "language" | "background"; type SettingsPage = Exclude; const tabIcons: Record = { + background: ServerCog, ownership: KeyRound, appearance: Palette, capabilities: ServerCog, @@ -85,6 +87,7 @@ export function WorkspaceSettingsPage({ { key: "conversation", label: locale === "zh-CN" ? "对话与模型" : "Conversations and models" }, { key: "provider", label: t("settings.modelProvider") }, { key: "capabilities", label: t("settings.globalCapabilities") }, + { key: "background", label: locale === "zh-CN" ? "后台服务" : "Background services" }, { key: "lark", label: "Lark" }, { key: "appearance", label: t("settings.appearance") }, { key: "language", label: t("settings.language") }, @@ -118,6 +121,7 @@ export function WorkspaceSettingsPage({ }, ]; const headings: Record = { + background: {title: locale === "zh-CN" ? "后台服务" : "Background services"}, conversation: {title: locale === "zh-CN" ? "Goal 对话默认值" : "Goal conversation defaults"}, ownership: {title: t("ownership.title")}, appearance: { @@ -146,6 +150,7 @@ export function WorkspaceSettingsPage({ const bodyRef = useRef(null); useEffect(() => { if (bodyRef.current) bodyRef.current.scrollTop = 0; }, [scope, initialGoalId, tab]); const scopeHint = scope === "machine" ? { + background: "", steward: "", conversation: t("settings.effect.conversation"), provider: t("settings.effect.provider"), @@ -237,6 +242,7 @@ export function WorkspaceSettingsPage({ ) : null} + {tab === "background" ? navigate("machine", undefined, tab)} /> : null} {tab === "steward" ? ; -const tabs = ["appearance", "capabilities", "language", "lark", "machine", "provider", "steward", "conversation", "cadence", "ownership"] as const; +const tabs = ["appearance", "capabilities", "language", "lark", "machine", "provider", "steward", "conversation", "cadence", "ownership", "background"] as const; export function readSettingsRoute(url: URL): SettingsTarget | null { const scope = url.searchParams.get("settings_scope"); From 5fc2b29b63691726217add822e76d2fe11096d9e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:18:28 +0530 Subject: [PATCH 5/8] fix(settings): keep mobile background navigation readable Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../src/features/personal-workspace/personal-workspace.css | 4 ++++ .../features/personal-workspace/workspace-settings-page.tsx | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css index 64fab49516..774d127e85 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css @@ -1695,6 +1695,10 @@ button.personal-execution-chip:focus-visible { outline: 2px solid #0070f3; outli } .personal-workspace-shell[data-pw-theme="loopx"] .personal-workspace-sidebar, +@media (max-width: 560px) { + .personal-settings-sidebar { grid-template-columns: minmax(0, 1fr) minmax(0, 1.25fr); } + .personal-settings-mobile-category > span { display: none; } +} .personal-settings-page[data-pw-theme="loopx"] .personal-settings-sidebar { background: #ffffff; } .personal-workspace-shell[data-pw-theme="loopx"] .personal-channel-header { background: rgb(250 250 250 / 94%); box-shadow: none; } .personal-workspace-shell[data-pw-theme="loopx"] .personal-channel-title h1 { font-size: var(--text-heading); line-height: var(--leading-heading); font-weight: 600; letter-spacing: 0; } diff --git a/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx index b8ff4e5cfc..37eeb0e0f9 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx @@ -181,7 +181,7 @@ export function WorkspaceSettingsPage({ From 6720943dc895566712a4f0ac0ba7f9a2075d4f17 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:23:18 +0530 Subject: [PATCH 6/8] refactor(runtime): centralize Chat background lifecycle adapters Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/chat_server.py | 24 ++++++------------- loopx/extensions/background_services.py | 1 - loopx/presentation/background_services_api.py | 17 ++++++++++++- .../test_background_services_api.py | 1 - 4 files changed, 23 insertions(+), 20 deletions(-) diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 24c79f06f9..0f07e449aa 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -95,10 +95,7 @@ ) from .extensions.lark.goal_topic_connections import list_lark_apps from .extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService -from .extensions.background_services import BackgroundServices -from .presentation.background_services_api import ( - BACKGROUND_SERVICES_PATH, BackgroundServicesRequestMixin, register_chat_background_services, -) +from .presentation import background_services_api as background_api from .extensions.lark.cli_transport import CommandRunner from .extensions.runtime import ( default_extension_state_file, @@ -482,30 +479,23 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self.runtime_identity = release_runtime_identity() super().__init__(*args, **kwargs) self.completed_todo_pages = CompletedTodoPages() - self.background_services = BackgroundServices() - self.background_service_owners: set[str] = set() + background_api.initialize_chat_background_services(self) def server_close(self) -> None: - self.background_services.close() - if hasattr(self, "codex_account_service") and "codex_account_service" not in self.background_service_owners: - self.codex_account_service.close() + background_api.close_chat_background_services(self) if hasattr(self, "lark_app_setup_manager"): self.lark_app_setup_manager.close() - if hasattr(self, "manager_return_service") and "manager_return_service" not in self.background_service_owners: - self.manager_return_service.close() if hasattr(self, "conversation_transports"): self.conversation_transports.close() if hasattr(self, "delegation_wake_service"): self.delegation_wake_service.close() - if hasattr(self, "lark_goal_topic_runtime") and "lark_goal_topic_runtime" not in self.background_service_owners: - self.lark_goal_topic_runtime.close() if hasattr(self, "runtime_controller"): self.runtime_controller.close() super().server_close() class ChatRequestHandler( - BackgroundServicesRequestMixin, + background_api.BackgroundServicesRequestMixin, ChatModelSettingsRequestMixin, GoalSourceRequestMixin, GoalAcceptanceRequestMixin, @@ -1707,7 +1697,7 @@ def do_GET(self) -> None: } ) get_dispatch: dict[str, Callable[[], None]] = { - BACKGROUND_SERVICES_PATH: self._background_services_read, + background_api.BACKGROUND_SERVICES_PATH: self._background_services_read, "/api/chat/request-inventory": self._request_inventory_read, GOAL_SOURCE_LINKS_PATH: self._goal_source_links, "/api/chat/todo/detail": self._todo_detail, @@ -1772,7 +1762,7 @@ def do_POST(self) -> None: if self._dispatch_zcode_goal(path, apply=True): return post_dispatch: dict[str, Callable[[], None]] = { - BACKGROUND_SERVICES_PATH: self._background_services_recheck, + background_api.BACKGROUND_SERVICES_PATH: self._background_services_recheck, GOAL_SOURCE_LINKS_PATH: self._goal_source_links, RESEARCH_ANALYSIS_PATH: self._research_analysis_write, RESEARCH_ANALYSIS_PATH + "/cancel": self._research_analysis_write, @@ -2036,7 +2026,7 @@ def _wake_goal_context(session: dict[str, object]) -> dict[str, object]: try: if external_conversation_factories: server.conversation_transports.start(server) - register_chat_background_services(server) + background_api.register_chat_background_services(server) server.background_services.start() server.serve_forever() except KeyboardInterrupt: diff --git a/loopx/extensions/background_services.py b/loopx/extensions/background_services.py index 083044871a..7620927ddd 100644 --- a/loopx/extensions/background_services.py +++ b/loopx/extensions/background_services.py @@ -114,4 +114,3 @@ def close(self) -> None: except Exception: with self._state_lock: self._errors.add(service.service_id) - diff --git a/loopx/presentation/background_services_api.py b/loopx/presentation/background_services_api.py index 5d70ef9190..c07081d3d3 100644 --- a/loopx/presentation/background_services_api.py +++ b/loopx/presentation/background_services_api.py @@ -1,11 +1,16 @@ """Chat composition and loopback read/recheck for the common background host.""" from __future__ import annotations -from ..extensions.background_services import BackgroundService, BackgroundTask, worker_observation +from ..extensions.background_services import BackgroundService, BackgroundServices, BackgroundTask, worker_observation BACKGROUND_SERVICES_PATH = "/api/chat/background-services" +def initialize_chat_background_services(server) -> None: + server.background_services = BackgroundServices() + server.background_service_owners = set() + + def register_chat_background_services(server) -> None: accounts = server.codex_account_service returns = server.manager_return_service @@ -25,6 +30,16 @@ def register_chat_background_services(server) -> None: server.background_service_owners = {"codex_account_service", "manager_return_service", "lark_goal_topic_runtime"} +def close_chat_background_services(server) -> None: + server.background_services.close() + # Directly constructed HTTP fixtures/embedders can still supply old owner + # attributes without registering. Close them once through the same adapter. + for attribute in ("codex_account_service", "manager_return_service", "lark_goal_topic_runtime"): + owner = getattr(server, attribute, None) + if owner is not None and attribute not in server.background_service_owners: + owner.close() + + class BackgroundServicesRequestMixin: def _background_services_read(self): self._send_json(self.server.background_services.snapshot()) diff --git a/tests/presentation/test_background_services_api.py b/tests/presentation/test_background_services_api.py index 8866f091e8..2631ffb68f 100644 --- a/tests/presentation/test_background_services_api.py +++ b/tests/presentation/test_background_services_api.py @@ -150,4 +150,3 @@ def test_supervisor_restores_enabled_observer_without_a_status_poll_or_policy_ed time.sleep(.02) assert provider.observer_health(True)["state"] == "healthy" assert provider.path.read_bytes() == saved - From f140708c3fe482073af8c7e7193631d9b671a073 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:26:08 +0530 Subject: [PATCH 7/8] fix(settings): keep responsive rules outside grouped selectors Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../src/features/personal-workspace/personal-workspace.css | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css index 774d127e85..68a90578a8 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css @@ -1694,11 +1694,11 @@ button.personal-execution-chip:focus-visible { outline: 2px solid #0070f3; outli --pw-caution: #f5a623; } -.personal-workspace-shell[data-pw-theme="loopx"] .personal-workspace-sidebar, @media (max-width: 560px) { .personal-settings-sidebar { grid-template-columns: minmax(0, 1fr) minmax(0, 1.25fr); } .personal-settings-mobile-category > span { display: none; } } +.personal-workspace-shell[data-pw-theme="loopx"] .personal-workspace-sidebar, .personal-settings-page[data-pw-theme="loopx"] .personal-settings-sidebar { background: #ffffff; } .personal-workspace-shell[data-pw-theme="loopx"] .personal-channel-header { background: rgb(250 250 250 / 94%); box-shadow: none; } .personal-workspace-shell[data-pw-theme="loopx"] .personal-channel-title h1 { font-size: var(--text-heading); line-height: var(--leading-heading); font-weight: 600; letter-spacing: 0; } From 063b361e62ea11382e1ccea76de35f662b5e4e01 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:32:07 +0530 Subject: [PATCH 8/8] fix(runtime): supervise the background host from the existing HTTP pump Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/extensions.md | 4 ++- loopx/chat_server.py | 3 +- loopx/extensions/background_services.py | 11 +++++++ loopx/presentation/background_services_api.py | 16 ++++++--- .../test_background_services_api.py | 33 +++++++++++++++++++ 5 files changed, 60 insertions(+), 7 deletions(-) diff --git a/docs/reference/extensions.md b/docs/reference/extensions.md index 36cd4a4608..1f9aad11af 100644 --- a/docs/reference/extensions.md +++ b/docs/reference/extensions.md @@ -50,7 +50,9 @@ The Chat process registers these four loops with the process-scoped startup and every 15 seconds, isolates owner failures, and fences shutdown before closing registered services. There is no additional saved enable switch, job queue, system daemon or permission grant. The App/OS still supervises the -Chat process. An enabled native observer that stopped is re-created from its +Chat process; its existing HTTP pump restores an unexpectedly exited host +thread with a cheap, idempotent liveness hook. Recheck also restores that host, +and never activates it from a passive read. An enabled native observer that stopped is re-created from its current saved policy; a live scan or a normally completed one-shot is not replayed. Settings → **Background services** reads the same content-free projection as diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 0f07e449aa..9b9f46f571 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -445,7 +445,7 @@ def resolve_lark_cli_for_runtime( ) -class ChatHTTPServer(ThreadingHTTPServer): +class ChatHTTPServer(background_api.BackgroundServicesServerMixin, ThreadingHTTPServer): allow_reuse_address = True registry_path: Path @@ -479,7 +479,6 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self.runtime_identity = release_runtime_identity() super().__init__(*args, **kwargs) self.completed_todo_pages = CompletedTodoPages() - background_api.initialize_chat_background_services(self) def server_close(self) -> None: background_api.close_chat_background_services(self) diff --git a/loopx/extensions/background_services.py b/loopx/extensions/background_services.py index 7620927ddd..c5ffd3811a 100644 --- a/loopx/extensions/background_services.py +++ b/loopx/extensions/background_services.py @@ -61,6 +61,17 @@ def start(self) -> None: self._started = True self._supervisor.start() + def supervise(self) -> None: + """Cheap process-pump hook; it cannot activate an unstarted host.""" + with self._state_lock: + started = self._started + if started: + self._supervisor.start() + + def recheck(self) -> None: + self.start() + self.reconcile() + def _run(self) -> None: self._supervisor.repeat(self.reconcile, interval_seconds=15) diff --git a/loopx/presentation/background_services_api.py b/loopx/presentation/background_services_api.py index c07081d3d3..f9c8778339 100644 --- a/loopx/presentation/background_services_api.py +++ b/loopx/presentation/background_services_api.py @@ -6,9 +6,17 @@ BACKGROUND_SERVICES_PATH = "/api/chat/background-services" -def initialize_chat_background_services(server) -> None: - server.background_services = BackgroundServices() - server.background_service_owners = set() +class BackgroundServicesServerMixin: + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.background_services = BackgroundServices() + self.background_service_owners = set() + + def service_actions(self): + # The existing HTTP pump can restore an unexpectedly exited host + # thread. No provider IO, scheduling or process restart runs here. + self.background_services.supervise() + super().service_actions() def register_chat_background_services(server) -> None: @@ -51,5 +59,5 @@ def _background_services_recheck(self): valid = False if not valid: return self._send_json({"ok": False, "code": "invalid_request"}, status=400) - self.server.background_services.reconcile() + self.server.background_services.recheck() self._send_json(self.server.background_services.snapshot()) diff --git a/tests/presentation/test_background_services_api.py b/tests/presentation/test_background_services_api.py index 2631ffb68f..42d25da531 100644 --- a/tests/presentation/test_background_services_api.py +++ b/tests/presentation/test_background_services_api.py @@ -86,6 +86,7 @@ def test_http_recheck_starts_existing_owners_once_and_preserves_disabled_feature time.sleep(.02) assert tasks["manager-returns"]["state"] == "healthy" assert tasks["manager-executions"]["state"] == "healthy" + assert result["supervisor"]["state"] in {"healthy", "checking"}, "Recheck must restore the host as well" request(server, "POST", {"operation": "recheck"}) assert server.manager_return_service.thread is first assert not provider.root.exists() and not server.lark_goal_topic_runtime._delivery_thread @@ -150,3 +151,35 @@ def test_supervisor_restores_enabled_observer_without_a_status_poll_or_policy_ed time.sleep(.02) assert provider.observer_health(True)["state"] == "healthy" assert provider.path.read_bytes() == saved + + +def test_http_pump_restores_an_unexpectedly_exited_supervisor_without_a_status_poll(host, monkeypatch): + server, _, _ = host + runtime = server.background_services + original_target, original_reconcile = runtime._supervisor.target, runtime.reconcile + exited, recovered = threading.Event(), threading.Event() + original_hook = threading.excepthook + + def expected_exit(args): + if args.thread.name == "loopx-background-services" and args.exc_type is SystemExit: + exited.set() + else: + original_hook(args) + + def target(): + if not exited.is_set(): + raise SystemExit("synthetic host exit") + original_target() + + def reconcile(): + original_reconcile() + recovered.set() + + monkeypatch.setattr(threading, "excepthook", expected_exit) + monkeypatch.setattr(runtime._supervisor, "target", target) + monkeypatch.setattr(runtime, "reconcile", reconcile) + runtime.start() + assert exited.wait(2) + # Only the real server's service_actions pump restores this thread. + assert recovered.wait(3) + assert runtime._supervisor.thread.is_alive()