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/personal-workspace.css b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css index 64fab49516..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,6 +1694,10 @@ button.personal-execution-chip:focus-visible { outline: 2px solid #0070f3; outli --pw-caution: #f5a623; } +@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; } 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..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 @@ -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"), @@ -176,7 +181,7 @@ export function WorkspaceSettingsPage({ @@ -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"); diff --git a/docs/reference/extensions.md b/docs/reference/extensions.md index 8cc1388b5b..1f9aad11af 100644 --- a/docs/reference/extensions.md +++ b/docs/reference/extensions.md @@ -25,16 +25,49 @@ 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 +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 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 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; 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 +`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 diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 94253626e6..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): @@ -609,14 +610,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, progress=self._return_worker.checkpoint) def run_executions(self) -> None: # Native previews may wait on a subprocess. Keep result publication @@ -625,12 +626,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/chat_server.py b/loopx/chat_server.py index a737c61852..9b9f46f571 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -95,6 +95,7 @@ ) from .extensions.lark.goal_topic_connections import list_lark_apps from .extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService +from .presentation import background_services_api as background_api from .extensions.lark.cli_transport import CommandRunner from .extensions.runtime import ( default_extension_state_file, @@ -444,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 @@ -480,24 +481,20 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self.completed_todo_pages = CompletedTodoPages() def server_close(self) -> None: - if hasattr(self, "codex_account_service"): - 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"): - 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"): - self.lark_goal_topic_runtime.close() if hasattr(self, "runtime_controller"): self.runtime_controller.close() super().server_close() class ChatRequestHandler( + background_api.BackgroundServicesRequestMixin, ChatModelSettingsRequestMixin, GoalSourceRequestMixin, GoalAcceptanceRequestMixin, @@ -1699,6 +1696,7 @@ def do_GET(self) -> None: } ) get_dispatch: dict[str, Callable[[], None]] = { + 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, @@ -1763,6 +1761,7 @@ def do_POST(self) -> None: if self._dispatch_zcode_goal(path, apply=True): return post_dispatch: dict[str, Callable[[], None]] = { + 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, @@ -1904,7 +1903,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 +1996,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 +2025,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() + background_api.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 29fe638fb2..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: @@ -52,6 +96,43 @@ 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. + 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: + 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 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) + def close(self, *, timeout: float = 2) -> None: self.stop.set() thread = self.thread diff --git a/loopx/extensions/background_services.py b/loopx/extensions/background_services.py new file mode 100644 index 0000000000..c5ffd3811a --- /dev/null +++ b/loopx/extensions/background_services.py @@ -0,0 +1,127 @@ +"""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 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) + + 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 aa83896ef4..b29702089f 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -60,23 +60,23 @@ 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.""" 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 = 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 +100,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 +175,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 +567,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/loopx/presentation/background_services_api.py b/loopx/presentation/background_services_api.py new file mode 100644 index 0000000000..f9c8778339 --- /dev/null +++ b/loopx/presentation/background_services_api.py @@ -0,0 +1,63 @@ +"""Chat composition and loopback read/recheck for the common background host.""" +from __future__ import annotations + +from ..extensions.background_services import BackgroundService, BackgroundServices, BackgroundTask, worker_observation + +BACKGROUND_SERVICES_PATH = "/api/chat/background-services" + + +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: + 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"} + + +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()) + + 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.recheck() + 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/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/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_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/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..42d25da531 --- /dev/null +++ b/tests/presentation/test_background_services_api.py @@ -0,0 +1,185 @@ +"""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" + 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 + 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 + + +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() 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" 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",