From 364967ff094d04b969a68c9dd52b9e6dc1f6e3db Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:20:12 +0800 Subject: [PATCH 1/4] feat(lark): notify owner tasks from the steward service Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- apps/presentation/dashboard/src/data/chat.ts | 13 + .../private-conversation-panel.tsx | 17 +- loopx/capabilities/manager_context/README.md | 36 +++ loopx/chat_server.py | 1 + .../control_plane/effect_runtime_handlers.ts | 3 + .../presentation/owner_attention.ts | 133 +++++++++ .../lark/goal_topic_runtime_service.py | 16 +- loopx/extensions/lark/owner_attention.py | 238 ++++++++++++++++ .../lark/private_conversation_api.py | 14 + .../extensions/lark/private_conversations.py | 2 + .../control_plane_ts/owner_attention.test.ts | 71 +++++ tests/test_lark_owner_attention.py | 257 ++++++++++++++++++ 12 files changed, 799 insertions(+), 2 deletions(-) create mode 100644 loopx/control_plane/presentation/owner_attention.ts create mode 100644 loopx/extensions/lark/owner_attention.py create mode 100644 tests/control_plane_ts/owner_attention.test.ts create mode 100644 tests/test_lark_owner_attention.py diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 161c8c5991..62fd4f375e 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -2504,6 +2504,12 @@ const privateConversationSchema = z.object({ goal_scope: z.enum(["all_registered", "selected"]).nullable().optional(), audience: z.enum(["owner", "group"]).default("owner"), group_count: z.number().int().default(0), pending_count: z.number().int(), recovery_count: z.number().int(), + owner_attention: z.discriminatedUnion("available", [ + z.object({available: z.literal(true), enabled: z.boolean(), revision: z.number().int(), + last_checked_at: z.number().nullable(), error_code: z.string().nullable(), + pending_count: z.number().int(), unconfirmed_count: z.number().int(), failed_count: z.number().int()}), + z.object({available: z.literal(false), error_code: z.string()}), + ]).optional(), agent_candidates: z.array(privateAgentSessionSchema).default([]), agent_targets: z.array(privateAgentTargetSchema).default([]), }); const privateConversationStopRecoverySchema = z.object({binding_id: z.string(), project_ref: z.string(), @@ -2535,6 +2541,13 @@ export async function changePrivateAgentTarget(bindingId: string, revision: numb })); } +export async function changeOwnerNotifications(bindingId: string, bindingRevision: number, revision: number, enabled: boolean) { + return privateConversationsSchema.parse(await requestJson("/api/chat/lark/private-conversations/owner-notifications", { + method: "POST", headers: {"Content-Type": "application/json"}, + body: JSON.stringify({binding_id: bindingId, binding_revision: bindingRevision, revision, enabled}), + })); +} + // Display validation only; finding status semantics remain owned by Explore. const exploreResultPageSchema = z.object({ diff --git a/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx index ad5e2b4bde..17cb5f20b9 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx @@ -1,6 +1,6 @@ import {useEffect, useState} from "react"; import {connectPrivateConversation, disconnectPrivateConversation, fetchPrivateConversations, - changePrivateAgentTarget, fetchChatProjects, fetchChatCapabilities, fetchLarkApps, fetchLarkGroupChats, type PrivateConversation, + changePrivateAgentTarget, changeOwnerNotifications, fetchChatProjects, fetchChatCapabilities, fetchLarkApps, fetchLarkGroupChats, type PrivateConversation, type ChatProject, type LarkApp, type PrivateConversationStopRecovery} from "../../data/chat"; import {useWorkspaceI18n} from "./i18n"; import "./private-conversation.css"; @@ -103,6 +103,21 @@ export function PrivateConversationPanel() { {row.audience === "group" ?

{zh ? `群话题助手 · 已授权 ${row.group_count} 个群 · 独立工作区隔离` : `Group topic assistant · ${row.group_count} authorized groups · Isolated workspace`}

: null}

{row.executor_endpoint_id} · {zh ? "监听状态" : "Listener"}: {listenerLabel(row.listener_status)}

{zh ? `待处理或回复：${row.pending_count}` : `Pending execution or reply: ${row.pending_count}`}

+ {row.context_kind === "steward" && row.audience === "owner" ?
+ {row.owner_attention?.available ? <> + +

{zh ? "开启后，管家服务自动通知授权范围内新增或实质变更的用户待办，发到原本人私聊。首次开启保留现有待办，不补发历史通知。关闭后暂停发送，保留原回执。" : "The steward service sends new or materially changed owner tasks within its scope to the original owner private Chat. First enable keeps existing tasks without a historical burst. Disabling pauses delivery and retains receipts."}

+ {row.owner_attention.enabled ?

{row.owner_attention.error_code || row.owner_attention.failed_count > 0 + ? (zh ? "通知暂未完成。请检查此 App 登录、授权工作与原私聊；原待办和回执仍保留，发送结果不明的通知不会重发。" : "Notifications need attention. Check this App login, authorized work and original Chat. Tasks and receipts are retained; ambiguous sends are not repeated.") + : row.owner_attention.last_checked_at === null ? (zh ? "等待首次后台检查" : "Waiting for the first background check") + : Date.now() / 1000 - row.owner_attention.last_checked_at > 120 ? (zh ? "后台检查尚未恢复，请确认管家服务正在运行。" : "Background checks have not resumed. Check that the steward service is running.") + : (zh ? `后台自动检查中 · 待发 ${row.owner_attention.pending_count}` : `Background checks active · ${row.owner_attention.pending_count} pending`)}

: null} + {row.owner_attention.unconfirmed_count > 0 ?

{zh ? "有通知发送结果尚未确认；服务只恢复原回执，不重复发送。" : "A notification receipt is unconfirmed; the service recovers the original receipt without resending."}

: null} + :

{zh ? "自动通知状态尚未确认，请刷新后重试。" : "Automatic notification state is unavailable. Refresh to retry."}

} +
: null} {row.recovery_count > 0 ?

{zh ? "存在尚未确认的发送回执。服务会读取原回执恢复；不要重新发送同一任务。检查 App 登录、权限和原会话后刷新状态。" : "A send receipt is unconfirmed. The service reads the original receipt to recover; avoid resending the same task. Check this App login, permissions and original Session, then refresh status."}

: null} {!row.context_available ?

{zh ? "工作区授权已失效；请恢复原工作区或重新选择。旧会话不会移到其它工作区。" : "The workspace grant is unavailable. Restore the original workspace or select a new one; the old Session will not move."}

: null} {row.context_kind === "project" && row.audience !== "group" ? : null} diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index c5d5e0bace..1633c217cd 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -153,6 +153,42 @@ worker planning workflow; do not overwrite the original receipt. This feature adds no periodic automation, forced wakeup, Todo priority or protected-operation permission. Existing private inbox records are retained when delivery is revoked. +## Automatic owner task notifications + +An existing owner-only private steward can enable **Automatically notify new +owner tasks** in Settings → Lark. This is an explicit, reversible notification +setting; existing installations stay off until enabled. First enable baselines +the current tasks without sending a historical burst. The ordinary steward +service then discovers new and materially changed `user_action` / `user_gate` +Todos in its current Goal scope, including subsequently registered Goals when +the binding already authorizes all registered work. No Codex heartbeat, model +Turn, hidden Goal, delegated worker or additional event consumer is involved. +This notice grants no execution or decision authority and does not complete a +Todo. Worker-result returns keep their separate existing delivery contract. + +Discovery reads every authorized Goal through the native Todo reader and +restores exact source text before preparing a notice. A failed or incomplete +read retains prior work; it cannot infer task removal. Completed, withdrawn, +superseded or out-of-scope sources cancel unsent notices. A local typed planner +owns subject identity, meaningful-content changes and reopening generations; +this is a local notification journal, not another Todo authority. + +The existing four-worker outbound pool reserves at most one slot for bounded +notification reconciliation. Each durable per-binding outbox has a process +fence. Before each send, Core rechecks the exact steward, App, owner, current +scope and original private-message provenance. Settings cannot supply another +destination. A frozen intent precedes the provider write; the independent +readback must match the exact message, App, private Chat and complete text. +Restart recovers an existing locator by reading it. A create without a returned +locator remains unconfirmed and is never blindly repeated; only a proven +pre-spawn failure permits up to three bounded attempts. Oversized complete +notices fail visibly instead of silently truncating decision terms. + +The setting and content-free pending/unconfirmed/failure health are read back +in the same settings panel. Disable remains local even when provider auth is +down, pauses delivery and retains all frozen receipts. Re-enable preserves +those receipts and discovers changes since the previous successful snapshot. + ## Scoped project conversations Local Goal Chat reuses this inbox and return path for members registered in that diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 5d1b8d70b3..9ca8dd9121 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -1586,6 +1586,7 @@ def do_POST(self) -> None: CHAT_LARK_CONNECTIONS_PATH: self._lark_connect, PRIVATE_CONVERSATIONS_PATH: self._private_conversation_connect, PRIVATE_CONVERSATIONS_PATH + "/agent-targets": self._private_conversation_agent_target, + PRIVATE_CONVERSATIONS_PATH + "/owner-notifications": self._private_conversation_owner_notifications, **self._configuration_post_routes(), **self._ssh_source_post_routes(), } diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index b4ccd3c6c7..cb0af7c041 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -457,6 +457,9 @@ export function createEffectRuntimeHandlers( ["presentation.decision_notice.project", lazyHandler(() => import("./presentation/decision_notice.ts"), ({projectDecisionNotice}) => projectDecisionNotice)], ["presentation.decision_notice.validate_references", lazyHandler(() => import("./presentation/decision_notice.ts"), ({validateDecisionNoticeReferences}) => validateDecisionNoticeReferences)], ["presentation.goal_attention.project", lazyHandler(() => import("./presentation/goal_attention.ts"), ({projectGoalAttention}) => projectGoalAttention)], + ["presentation.owner_attention.plan", lazyHandler(() => import("./presentation/owner_attention.ts"), ({planOwnerAttention}) => planOwnerAttention)], + ["presentation.owner_attention.observe", lazyHandler(() => import("./presentation/owner_attention.ts"), ({observeOwnerAttention}) => observeOwnerAttention)], + ["presentation.owner_attention.health", lazyHandler(() => import("./presentation/owner_attention.ts"), ({ownerAttentionHealth}) => ownerAttentionHealth)], ["presentation.goal_attention.bound", lazyHandler(() => import("./presentation/goal_attention.ts"), ({boundGoalAttention}) => boundGoalAttention)], ["presentation.goal_attention.reconcile", lazyHandler(() => import("./presentation/goal_attention.ts"), ({reconcileGoalAttention}) => reconcileGoalAttention)], ["presentation.action_review_plan.compile", lazyHandler(() => import("./presentation/action_review_plan.ts"), ({compileActionReviewPlan}) => (params) => diff --git a/loopx/control_plane/presentation/owner_attention.ts b/loopx/control_plane/presentation/owner_attention.ts new file mode 100644 index 0000000000..fd788cf6f7 --- /dev/null +++ b/loopx/control_plane/presentation/owner_attention.ts @@ -0,0 +1,133 @@ +/** Notification discovery owns no Todo lifecycle or execution authority. + * Complete native snapshots select subjects; provider journals prove delivery. */ +import type {JsonObject} from "../effect_program.ts"; +import {requireInteger, requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; +import {canonicalAuthoritySha256 as digest} from "../coordination/authority_store_codec.ts"; + +const records = (value: unknown, name: string): JsonObject[] => { + if (!Array.isArray(value)) throw new TypeError(`${name} must be a complete array`); + return value.map(row => requireJsonObject(row, name)); +}; +const active = (row: JsonObject) => row.role === "user" && row.done !== true + && !row.superseded_by && ["open", "blocked"].includes(String(row.status)) + && ["user_action", "user_gate"].includes(String(row.task_class)); + +/** Local provider-journal vocabulary; never canonical Todo/Run status. */ +function readState(raw: unknown): JsonObject { + const state = requireJsonObject(raw, "state"); + if (state.schema_version !== "lark_owner_attention_v0" || typeof state.enabled !== "boolean" + || typeof state.initialized !== "boolean" || requireInteger(state.revision, "revision") < 0) + throw new Error("owner notification state is unavailable"); + records(state.subjects, "subjects"); + for (const row of records(state.outbox, "outbox")) { + if (!["pending", "sending", "unconfirmed", "delivered", "failed", "superseded"].includes(String(row.status))) + throw new Error("unknown notification receipt state"); + } + return state; +} + +export function ownerAttentionHealth(input: JsonObject): JsonObject { + const state = readState(input.state); + const rows = records(state.outbox, "outbox"); + return {available: true, enabled: state.enabled, revision: state.revision, + last_checked_at: state.last_checked_at ?? null, error_code: state.error_code ?? null, + pending_count: rows.filter(row => row.status === "pending").length, + unconfirmed_count: rows.filter(row => row.status === "sending" || row.status === "unconfirmed").length, + failed_count: rows.filter(row => row.status === "failed").length}; +} + +export function planOwnerAttention(input: JsonObject): JsonObject { + const state = readState(input.state); + if (input.source_complete !== true) throw new Error("owner attention source is incomplete"); + const previous = records(state.subjects, "subjects"); + const outbox = records(state.outbox, "outbox").map(row => ({...row})); + const current = new Map(); + for (const goal of records(input.goals, "goals")) { + const goalId = requireNonEmptyString(goal.goal_id, "goal_id"); + for (const todo of records(goal.todos, "todos")) { + if (!active(todo)) continue; + if (todo.goal_id != null && todo.goal_id !== goalId) throw new Error("foreign owner Todo"); + const todoId = requireNonEmptyString(todo.todo_id, "todo_id"); + const text = requireNonEmptyString(todo.text, "todo.text"); + const terms: JsonObject = {text}; + for (const key of ["note", "reason", "evidence", "decision_scope", "required_decision_scopes"]) { + if (todo[key] != null && todo[key] !== "") terms[key] = todo[key]; + } + const key = `${goalId}/${todoId}`; + if (current.has(key)) throw new Error("duplicate owner Todo"); + current.set(key, {key, goal_id: goalId, todo_id: todoId, terms, digest: digest(terms), active: true}); + } + } + const subjects: JsonObject[] = [...current.values()].map(row => { + const old = previous.find(item => item.key === row.key); + const changed = !old || old.active !== true || old.digest !== row.digest; + const generation = Number(old?.generation ?? 0) + Number(changed); + if (state.initialized === true && changed) { + const id = digest([row.key, generation, row.digest]); + const terms = row.terms as JsonObject; + const body = [`待你处理 · ${row.goal_id}`, String(terms.text), + ...["note", "reason", "evidence", "decision_scope", "required_decision_scopes"].flatMap(key => + terms[key] == null ? [] : [`${({note: "说明", reason: "原因", evidence: "依据", + decision_scope: "决定范围", required_decision_scopes: "决定范围"} as Record)[key]}：${ + typeof terms[key] === "string" ? terms[key] : JSON.stringify(terms[key])}`]), + "可在原管家对话中查看或答复这项待办。"].join("\n\n"); + outbox.push({id, key: row.key, generation, digest: row.digest, goal_id: row.goal_id, + todo_id: row.todo_id, body, status: Buffer.byteLength(body, "utf8") > 20000 ? "failed" : "pending", + error_code: Buffer.byteLength(body, "utf8") > 20000 ? "notice_content_too_large" : null, + attempts: 0, next_attempt_at: 0}); + } + const {terms: _terms, ...subject} = row; + return {...subject, generation}; + }); + for (const row of previous) { + if (!current.has(String(row.key))) subjects.push({...row, active: false}); + } + for (const row of outbox) { + const subject = subjects.find(item => item.key === row.key); + const obsolete = subject?.active !== true || subject.digest !== row.digest || subject.generation !== row.generation; + // A crash after starting a write is ambiguous even without a locator. + if (row.status === "sending") {row.status = "unconfirmed"; row.error_code = "send_outcome_unknown";} + if (obsolete && ["pending", "failed"].includes(String(row.status))) row.status = "superseded"; + row.source_current = !obsolete; + } + return {...state, initialized: true, subjects, outbox}; +} + +/** Fold provider observations, preserving frozen intent and ambiguous writes. */ +export function observeOwnerAttention(input: JsonObject): JsonObject { + const state = readState(input.state); + const id = requireNonEmptyString(input.id, "id"); + const event = requireNonEmptyString(input.event, "event"); + const outbox = records(state.outbox, "outbox").map(row => ({...row})); + const row = outbox.find(item => item.id === id); + if (!row) throw new Error("unknown owner notice"); + const now = Number(input.now); + if (!Number.isFinite(now) || now < 0) throw new Error("invalid notice observation time"); + if (event === "write_started") { + if (row.status !== "pending" || row.source_current !== true || Number(row.next_attempt_at) > now) + throw new Error("notice is not ready for delivery"); + row.intent = requireJsonObject(input.intent, "intent"); + row.status = "sending"; row.attempts = Number(row.attempts) + 1; row.error_code = null; + } else if (event === "spawn_failed") { + if (row.status !== "sending") throw new Error("notice write was not started"); + row.status = Number(row.attempts) >= 3 ? "failed" : "pending"; + row.error_code = "provider_spawn_failed"; + row.next_attempt_at = now + 30 * Number(row.attempts); + } else if (event === "write_observed") { + if (row.status !== "sending") throw new Error("notice write was not started"); + row.status = "unconfirmed"; + row.message_id = typeof input.message_id === "string" && input.message_id ? input.message_id : null; + row.error_code = row.message_id ? "readback_pending" : "send_outcome_unknown"; + row.next_attempt_at = now; + } else if (event === "readback_verified" || event === "readback_pending") { + if (row.status !== "unconfirmed" || !row.message_id) throw new Error("notice has no provider locator"); + row.status = event === "readback_verified" ? "delivered" : "unconfirmed"; + row.error_code = event === "readback_verified" ? null : "readback_pending"; + row.next_attempt_at = now + 60; + if (event === "readback_verified") { + row.readback = requireJsonObject(input.readback, "readback"); + row.verified_at = now; + } + } else throw new Error("unknown notice observation"); + return {...state, outbox}; +} diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index d4bb30cbf2..e4dbef48f0 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -9,6 +9,7 @@ import logging from pathlib import Path import threading +import time from typing import Any from ...chat_manager import MANAGER_AGENT_OBJECTIVE @@ -82,8 +83,17 @@ def _reconcile_private(self) -> None: return pending: dict[Path, Future[int]] = {} attempted: dict[Path, int] = {} + attention: Future[int] | None = None + attention_due = 0.0 with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool: while not self._closed.is_set(): + if attention is not None and attention.done(): + try: + attention.result() + except Exception: + logging.getLogger(__name__).warning("Owner notification reconciliation is pending") + attention = None + attention_due = time.monotonic() + 30 for path, future in list(pending.items()): if not future.done(): continue @@ -95,13 +105,17 @@ def _reconcile_private(self) -> None: try: paths = transport.pending_delivery_paths() attempted = {path: count for path, count in attempted.items() if path in paths} + owner_attention = getattr(transport, "owner_attention", None) + attention_ready = owner_attention is not None and attention is None and time.monotonic() >= attention_due for path in sorted(paths, key=lambda path: attempted.get(path, 0)): - if len(pending) >= 4 or self._closed.is_set(): + if len(pending) + int(attention is not None or attention_ready) >= 4 or self._closed.is_set(): break if path in pending: continue pending[path] = pool.submit(transport.reconcile_request, path) attempted[path] = attempted.get(path, 0) + 1 + if attention_ready and len(pending) < 4 and not self._closed.is_set(): + attention = pool.submit(owner_attention.reconcile) except Exception: logging.getLogger(__name__).warning("Private Chat delivery discovery is pending") self._closed.wait(1) diff --git a/loopx/extensions/lark/owner_attention.py b/loopx/extensions/lark/owner_attention.py new file mode 100644 index 0000000000..50654cd71e --- /dev/null +++ b/loopx/extensions/lark/owner_attention.py @@ -0,0 +1,238 @@ +"""Automatic owner notices on an existing Core-authorized private steward. + +Native Todo readers and the typed planner own discovery. This provider adapter +stores only settings, frozen intents and independently verified delivery facts. +It starts no listener, model Turn, delegated task or heartbeat. +""" +from __future__ import annotations + +import hashlib +import json +import re +import time +from pathlib import Path +from typing import Any + +from ...chat_store import _read_json +from ...control_plane.effect_runtime import effect_runtime_result +from ...control_plane.todos.list_readback import list_goal_todos +from ...file_lock import exclusive_file_lock, try_exclusive_file_lock, write_json_atomic +from .goal_channel_transport import call, json_payload, lark_args +from .inbox_reply import _message +from .outbound import lark_readback_matches_outbound, present_lark_conversation_text + + +class LarkOwnerAttention: + def __init__(self, transport: Any) -> None: + self.transport = transport + self.root = transport.core.controller.store.root / "lark-owner-attention" + + def _path(self, binding_id: str) -> Path: + if not re.fullmatch(r"[a-f0-9]{24}", binding_id): + raise ValueError("select an existing private steward") + return self.root / f"{binding_id}.json" + + @staticmethod + def _empty() -> dict[str, Any]: + return {"schema_version": "lark_owner_attention_v0", "revision": 0, "enabled": False, + "initialized": False, "subjects": [], "outbox": [], "last_checked_at": None, + "error_code": None} + + def _read(self, path: Path) -> dict[str, Any]: + state = _read_json(path) if path.exists() else self._empty() + if state.get("schema_version") != "lark_owner_attention_v0": + raise ValueError("owner notification state is unavailable") + return state + + @staticmethod + def _write(path: Path, state: dict[str, Any]) -> None: + # Cold receipts retain full intent/proof without feeding a growing + # history back through every planning RPC. Subject generations remain + # in the hot journal, so compaction cannot re-notify an unchanged task. + for row in state["outbox"]: + if row["status"] not in {"delivered", "superseded"}: + continue + receipt = path.parent / "receipts" / path.stem / f"{row['id']}.json" + if receipt.exists(): + saved = _read_json(receipt) + if any(saved.get(key) != row.get(key) for key in ("id", "key", "digest", "intent", "message_id", "status")): + raise ValueError("notification receipt identity changed") + else: + write_json_atomic(receipt, row, mode=0o600, directory_mode=0o700, fsync_directory=True) + state["outbox"] = [row for row in state["outbox"] if row["status"] not in {"delivered", "superseded"}] + write_json_atomic(path, state, mode=0o600, directory_mode=0o700, fsync_directory=True) + + def health(self, binding_id: str) -> dict[str, Any]: + try: + state = self._read(self._path(binding_id)) + return dict(effect_runtime_result("presentation.owner_attention.health", {"state": state})) + except (ValueError, OSError, KeyError, TypeError, RuntimeError): + return {"available": False, "error_code": "notification_state_unavailable"} + + def _binding(self, binding_id: str, expected_revision: int | None = None) -> dict[str, Any]: + current = self.transport.bindings.read() + if expected_revision is not None and current["revision"] != expected_revision: + raise ValueError("the connection changed; refresh before changing notifications") + row = next((row for row in current["bindings"] if row["binding_id"] == binding_id), None) + if (not row or row.get("enabled") is not True or row.get("context_kind") != "steward" + or row.get("audience", "owner") != "owner" or row.get("grant") != "portfolio_read"): + raise ValueError("notifications require an authorized owner private steward") + return row + + def _route(self, binding: dict[str, Any], saved: dict[str, Any] | None = None) -> dict[str, Any]: + candidates = [saved] if saved is not None else [ + _read_json(path) for path in sorted(self.transport.root.glob("*.json"), reverse=True) + if re.fullmatch(r"[a-f0-9]{24}", path.stem)] + for record in candidates: + if not isinstance(record, dict): + continue + event = record.get("event") or {} + if (record.get("binding_id") != binding["binding_id"] + or record.get("profile") != binding["transport_ref"] + or event.get("chat_type") != "p2p" or event.get("sender_type") != "user" + or not re.fullmatch(r"[a-f0-9]{24}", str(record.get("request_ref") or ""))): + continue + if not (self.transport.core.root / f"{record['request_ref']}.json").exists(): + continue + native = self.transport.core.read_request(record["request_ref"]) + if (native.get("binding_id") != binding["binding_id"] or native.get("source") != record.get("source") + or native.get("origin", "lark") != "lark" + or native.get("status") in {"prepared", "rejected"}): + continue + selected = self.transport.bindings.resolve(binding_id=binding["binding_id"], + **record["source"], include_observation=True) + if not self.transport._source_message(record, selected=selected, render_content=False): + continue + # Retain the exact canonical original source; never accept a chat + # ID or App supplied by the settings request or a Todo body. + return {**{key: record[key] for key in ("request_ref", "binding_id", "profile", "source")}, + "bot_app_id": selected["observation"]["bot_app_id"], + "event": {key: event[key] for key in ("message_id", "chat_id", "chat_type", "sender_type", "sender_id")}} + raise ValueError("send a message to this private steward before enabling notifications") + + def _snapshot(self, binding: dict[str, Any]) -> list[dict[str, Any]]: + controller = self.transport.core.controller + if controller.registry_path is None: + raise ValueError("the steward registry is unavailable") + goals = [] + for goal_id in self.transport.bindings.goal_scope_ids(binding): + args = dict(registry_path=controller.registry_path, goal_id=goal_id, + runtime_root_arg=str(controller.coordination_runtime_root), role="user") + listing = list_goal_todos(**args) + if listing.get("ok") is not True or listing.get("todo_count") != len(listing.get("todos", [])): + raise ValueError("the native Todo source is incomplete") + todos = [] + # Exact reads restore source text; a bounded summary never supplies + # notification terms. Read every listed user row without a prefix. + for row in listing["todos"]: + detail = list_goal_todos(**args, todo_id=row["todo_id"]) + if detail.get("ok") is not True or not detail.get("matched") or not isinstance(detail.get("todo"), dict): + raise ValueError("a native Todo changed while reading") + todos.append(detail["todo"]) + goals.append({"goal_id": goal_id, "todos": todos}) + return goals + + @staticmethod + def _plan(state: dict[str, Any], goals: list[dict[str, Any]]) -> dict[str, Any]: + return dict(effect_runtime_result("presentation.owner_attention.plan", + {"state": state, "goals": goals, "source_complete": True})) + + @staticmethod + def _observe(state: dict[str, Any], row: dict[str, Any], event: str, **facts: Any) -> dict[str, Any]: + return dict(effect_runtime_result("presentation.owner_attention.observe", + {"state": state, "id": row["id"], "event": event, "now": time.time(), **facts})) + + def configure(self, *, binding_id: str, binding_revision: int, revision: int, enabled: bool) -> None: + if (type(enabled) is not bool or type(revision) is not int or type(binding_revision) is not int): + raise ValueError("select the current notification settings") + path = self._path(binding_id) + with exclusive_file_lock(path, operation="configure_owner_notifications"): + binding = self._binding(binding_id, binding_revision) + state = self._read(path) + if state["revision"] != revision: + raise ValueError("notification settings changed; refresh first") + if enabled and not state["enabled"]: + # First enable baselines existing tasks, avoiding an unsolicited + # historical burst. Re-enable retains pending/unknown receipts. + state["route"] = self._route(binding, state.get("route")) + state = self._plan(state, self._snapshot(binding)) + state["error_code"] = None + state.update(enabled=enabled, revision=revision + 1) + self._write(path, state) + + def _verify(self, row: dict[str, Any], route: dict[str, Any], binding: dict[str, Any]) -> dict[str, Any] | None: + observed = self.transport.bindings.observe(route["profile"]) + if observed["provider_ref"] != binding["provider_ref"] or observed["operator_ref"] != binding["operator_ref"]: + return None + result = call(self.transport.runner, lark_args(cli_bin=self.transport.cli_bin, profile=route["profile"], + tail=["im", "+messages-mget", "--message-ids", row["message_id"], "--as", "bot", "--no-reactions", "--format", "json"])) + message = _message(json_payload(result), row["message_id"]) + sender = message.get("sender", {}) if message else {} + matches = bool(result.get("returncode") == 0 and message and not message.get("deleted") + and message.get("chat_id") == route["event"]["chat_id"] and sender.get("sender_type") == "app" + and sender.get("id") == observed["bot_app_id"] + and lark_readback_matches_outbound(outbound_text=row["intent"]["text"], message=message)) + return {"message_id": row["message_id"], "chat_id": message["chat_id"], "app_id": sender["id"], + "message_sha256": hashlib.sha256(json.dumps(message, sort_keys=True, ensure_ascii=False).encode()).hexdigest()} if matches else None + + def reconcile(self) -> int: + verified = 0 + # Each binding has one fenced outbox across processes. The caller uses + # the existing bounded outbound pool, never a new message consumer. + for path in sorted(self.root.glob("*.json")): + if not re.fullmatch(r"[a-f0-9]{24}", path.stem): + continue + with try_exclusive_file_lock(path, operation="deliver_owner_notifications") as acquired: + if not acquired: + continue + try: + state = self._read(path) + except (ValueError, OSError, KeyError, TypeError): + continue # one unreadable binding cannot starve other outboxes + if not state["enabled"]: + continue + try: + binding = self._binding(path.stem) + state = self._plan(state, self._snapshot(binding)) + state.update(last_checked_at=time.time(), error_code=None) + self._write(path, state) + ready = [row["id"] for row in state["outbox"] if row["status"] in {"pending", "unconfirmed"} + and float(row.get("next_attempt_at", 0)) <= time.time() + and (row["status"] == "pending" or row.get("message_id"))][:2] + for notice_id in ready: + # Withdrawals and scope revocations are checked again + # before each write, including after slow provider IO. + binding = self._binding(path.stem) + state = self._plan(state, self._snapshot(binding)) + self._write(path, state) + row = next((row for row in state["outbox"] if row["id"] == notice_id), None) + if row is None or row["status"] not in {"pending", "unconfirmed"}: + continue + route = self._route(binding, state["route"]) + if row["status"] == "pending": + text = present_lark_conversation_text(row["body"]) + intent = {"route": route, "text": text, "idempotency_key": "owner-attn-" + row["id"][:32]} + state = self._observe(state, row, "write_started", intent=intent) + self._write(path, state) # frozen before provider effect + result = call(self.transport.runner, lark_args(cli_bin=self.transport.cli_bin, profile=route["profile"], + tail=["im", "+messages-send", "--as", "bot", "--chat-id", route["event"]["chat_id"], + "--text", text, "--idempotency-key", intent["idempotency_key"], "--json"])) + payload = json_payload(result) + message_id = (payload.get("data") or {}).get("message_id") + state = self._observe(state, row, "spawn_failed" if result.get("spawn_failed") else "write_observed", + message_id=message_id) + self._write(path, state) # locator survives failed readback/restart + row = next(row for row in state["outbox"] if row["id"] == notice_id) + if row["status"] == "unconfirmed" and row.get("message_id"): + if row["intent"]["route"] != route: + raise ValueError("the original notification destination changed") + readback = self._verify(row, route, binding) + state = self._observe(state, row, "readback_verified" if readback else "readback_pending", readback=readback) + self._write(path, state) + verified += int(readback is not None) + except (ValueError, OSError, KeyError, TypeError, RuntimeError): + # Keep prior journals and unknown outcomes; no retry of an + # ambiguous create. Expose a safe actionable health state. + state["error_code"] = "source_or_delivery_unavailable" + self._write(path, state) + return verified diff --git a/loopx/extensions/lark/private_conversation_api.py b/loopx/extensions/lark/private_conversation_api.py index ae40b9c278..d705bab0c6 100644 --- a/loopx/extensions/lark/private_conversation_api.py +++ b/loopx/extensions/lark/private_conversation_api.py @@ -49,6 +49,8 @@ def _private_conversations(self) -> None: for item in bindings.agent_candidates(row["binding_id"])], "agent_targets": row.get("agent_targets", []), "listener_status": self._private_listener_status(row["transport_ref"], health), + **({"owner_attention": self.server.lark_private_conversations.owner_attention.health(row["binding_id"])} + if row["context_kind"] == "steward" and row.get("audience", "owner") == "owner" else {}), **deliveries.get(row["binding_id"], {"pending_count": 0, "recovery_count": 0})} for row in current["bindings"]], "stop_recoveries": [{**row, @@ -128,3 +130,15 @@ def _private_conversation_agent_target(self) -> None: self._send_error(str(exc), status=400) return self._private_conversations() + + def _private_conversation_owner_notifications(self) -> None: + try: + body = self._read_json() + if set(body) != {"binding_id", "binding_revision", "revision", "enabled"}: + raise ValueError("select the current private steward notification setting") + self.server.lark_private_conversations.owner_attention.configure(**body) + except (ValueError, OSError, KeyError, TypeError, RuntimeError): + # Source/provider errors may contain private paths or credentials. + self._send_error("Notification settings could not be changed. Refresh the connection and verify the original private Chat.", status=400) + return + self._private_conversations() diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 6a6f983194..2d9fcbe0c9 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -43,6 +43,8 @@ def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin: self.runtime_root, self.runner, self.cli_bin = runtime_root, runner, cli_bin self.reaction_feedback = reaction_feedback self.root = controller.store.root / "lark-private-deliveries" + from .owner_attention import LarkOwnerAttention + self.owner_attention = LarkOwnerAttention(self) def profiles(self) -> dict[str, dict[str, str]]: """Discover current App configuration, never authorize a conversation. diff --git a/tests/control_plane_ts/owner_attention.test.ts b/tests/control_plane_ts/owner_attention.test.ts new file mode 100644 index 0000000000..bd2841a01c --- /dev/null +++ b/tests/control_plane_ts/owner_attention.test.ts @@ -0,0 +1,71 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {observeOwnerAttention, ownerAttentionHealth, planOwnerAttention} from "../../loopx/control_plane/presentation/owner_attention.ts"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; + +const empty = () => ({schema_version: "lark_owner_attention_v0", revision: 0, enabled: true, initialized: false, subjects: [], outbox: []}); +const task = {todo_id: "todo_one", role: "user", status: "open", task_class: "user_action", text: "Inspect the preview"}; +const plan = (state: JsonObject, todos: JsonObject[], goalId = "alpha") => + planOwnerAttention({state, goals: [{goal_id: goalId, todos}], source_complete: true}); +const rows = (state: JsonObject) => state.outbox as JsonObject[]; + +test("first enable baselines history; later discovery is complete and stable across restart", () => { + const initial = plan(empty(), [task]); + assert.equal(rows(initial).length, 0); + const tasks = Array.from({length: 80}, (_, i) => ({...task, todo_id: `todo_${i}`})); + const discovered = plan(initial, [task, ...tasks]); + assert.equal(rows(discovered).length, 80, "no bounded prefix silently loses owner tasks"); + assert.deepEqual(plan(JSON.parse(JSON.stringify(discovered)), [task, ...tasks]), discovered); + assert.equal(rows(plan(discovered, [{...task, updated_at: "later"}, ...tasks])).length, 80); + assert.equal(rows(plan(discovered, [{...task, note: "New decision terms"}, ...tasks])).length, 81); +}); + +test("withdrawn or superseded tasks cancel unsent notices; reopening retains a new generation", () => { + const base = plan(empty(), []); + const newTask = plan(base, [task]); + const withdrawn = plan(newTask, [{...task, status: "done"}]); + assert.equal(rows(withdrawn)[0].status, "superseded"); + const reopened = plan(withdrawn, [task]); + assert.equal(rows(reopened).length, 2); + assert.notEqual(rows(reopened)[0].id, rows(reopened)[1].id); + assert.equal(rows(plan(base, [{...task, superseded_by: "todo_new"}, {...task, role: "agent"}])).length, 0); + assert.throws(() => planOwnerAttention({state: newTask, goals: [], source_complete: false})); + assert.throws(() => plan(base, [{...task, goal_id: "foreign"}])); + assert.throws(() => plan(base, [task, task])); +}); + +test("ambiguous create never returns to pending; locator recovers by independent read only", () => { + const state = plan(plan(empty(), []), [task]); + const id = rows(state)[0].id; + const observe = (s: JsonObject, event: string, extra = {}) => observeOwnerAttention({state: s, id, event, now: 100, ...extra}); + const started = observe(state, "write_started", {intent: {text: "frozen", route: {source: "original"}}}); + assert.equal(rows(plan(started, [task]))[0].status, "unconfirmed", "crash before saving a locator remains unknown"); + const unknown = observe(started, "write_observed"); + assert.equal(rows(unknown)[0].message_id, null); + assert.throws(() => observe(unknown, "write_started", {intent: {}})); + const located = observe(started, "write_observed", {message_id: "om_original"}); + const retry = observe(located, "readback_pending"); + assert.equal(rows(retry)[0].status, "unconfirmed"); + assert.equal(rows(retry)[0].message_id, "om_original"); + assert.equal(rows(observe(retry, "readback_verified", {readback: {message_id: "om_original"}}))[0].status, "delivered"); + assert.equal(rows(observe(started, "spawn_failed"))[0].status, "pending", "only proven no-effect failure can retry create"); +}); + +test("identical titles across Goals remain distinct; content is complete or explicitly failed", () => { + const base = plan(empty(), []); + const a = plan(base, [task]); + const b = planOwnerAttention({state: a, source_complete: true, + goals: [{goal_id: "alpha", todos: [task]}, {goal_id: "beta", todos: [task]}]}); + assert.equal(rows(b).length, 2); + assert.notEqual(rows(b)[0].id, rows(b)[1].id); + const long = plan(base, [{...task, text: "题".repeat(9000)}]); + assert.equal(rows(long)[0].status, "failed"); + assert.ok(String(rows(long)[0].body).includes("题".repeat(9000))); +}); + +test("unknown journal states cannot present healthy or authorize delivery", () => { + const broken = {...empty(), outbox: [{status: "new_unknown_state"}]}; + assert.throws(() => ownerAttentionHealth({state: broken})); + assert.throws(() => plan(broken, [task])); + assert.throws(() => ownerAttentionHealth({state: {...empty(), enabled: "true"}})); +}); diff --git a/tests/test_lark_owner_attention.py b/tests/test_lark_owner_attention.py new file mode 100644 index 0000000000..2ca4029aea --- /dev/null +++ b/tests/test_lark_owner_attention.py @@ -0,0 +1,257 @@ +"""Automatic owner notices through native sources and existing runtime workers.""" +import http.client +import json +import subprocess +import threading +import time +from types import SimpleNamespace + +import pytest + +from test_chat_ordinary_project import ordinary # noqa: F401 +from test_lark_private_conversations import connect +from loopx.chat_server import ChatHTTPServer, ChatRequestHandler +from loopx.extensions.lark.private_conversations import LarkPrivateConversations +from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService + + +def receipt(o): + return json.loads(next((o.attention.root / "receipts" / o.binding["binding_id"]).glob("*.json")).read_text()) + + +@pytest.fixture +def owner(ordinary, tmp_path): # noqa: F811 + store, runtime, provider, transport = connect(ordinary) + registry = runtime.registry_path + registry.write_text(json.dumps({"schema_version": 1, "goals": []})) + transport.bindings.configure(transport_ref="steward-app", project_ref=runtime.project_contexts.available()[0]["project_ref"], + executor_endpoint_id="codex", context_kind="steward") + event = provider.event("steward-app", "owner_original", "/status") + transport.admit("steward-app", event) + transport.reconcile() + binding = transport._binding("steward-app") + attention = transport.owner_attention + def configure(enabled=True, revision=None): + attention.configure(binding_id=binding["binding_id"], binding_revision=transport.bindings.read()["revision"], + revision=attention.health(binding["binding_id"])["revision"] if revision is None else revision, + enabled=enabled) + def tasks(rows, *, goal="alpha"): + path = tmp_path / f"{goal}.md" + path.write_text("# Goal\n\n## User Todo\n\n" + "".join( + f"- [{'x' if row.get('done') else ' '}] {row['text']}\n" + f" \n" + for row in rows) + "\n## Agent Todo\n") + data = json.loads(registry.read_text()) + data["goals"] = [row for row in data["goals"] if row["id"] != goal] + [ + {"id": goal, "repo": str(tmp_path), "state_file": path.name}] + registry.write_text(json.dumps(data)) + yield SimpleNamespace(store=store, runtime=runtime, provider=provider, transport=transport, binding=binding, + attention=attention, configure=configure, tasks=tasks, event=event, capture=ordinary[4]) + runtime.close() + + +def test_service_discovers_new_native_tasks_without_model_or_agent_poll(owner): + o = owner + o.tasks([{"id": "todo_old", "text": "Existing owner decision"}]) + o.configure() + before = len(o.provider.writes) + service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=o.store.root.parent, + runtime_controller=o.runtime, private_conversations=o.transport) + # The production outbound worker, no call to the notice reconciler and no + # synthetic inbound message or model Turn after creating the native Todo. + o.tasks([{"id": "todo_old", "text": "Existing owner decision"}, + {"id": "todo_new", "text": "Review preview " + "complete terms " * 100}]) + worker = threading.Thread(target=service._reconcile_private, daemon=True) + worker.start() + try: + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + if len(o.provider.writes) > before and o.attention.health(o.binding["binding_id"])["unconfirmed_count"] == 0: + break + time.sleep(.05) + assert len(o.provider.writes) == before + 1 + assert ("complete terms " * 100).rstrip() in o.provider.writes[-1][1], "exact source text survives summary bounds" + assert o.attention.reconcile() == 0 + assert len(o.provider.writes) == before + 1 + assert not o.capture.exists() and not o.store.list_sessions(), "no background model or hidden Goal" + row = receipt(o) + assert row["status"] == "delivered" and row["message_id"] in o.provider.messages + assert row["readback"]["message_id"] == row["message_id"] and row["readback"]["app_id"] == "cli_steward_app" + finally: + service._closed.set() + worker.join(10) + + +def test_restart_verifies_original_locator_and_never_resends(owner): + o = owner + o.configure() + o.tasks([{"id": "todo_new", "text": "Inspect a new preview"}]) + o.provider.verify_replies = False + before = len(o.provider.writes) + assert o.attention.reconcile() == 0 + assert o.attention.health(o.binding["binding_id"])["unconfirmed_count"] == 1 + restarted = LarkPrivateConversations(controller=o.runtime, runtime_root=o.store.root.parent, + runner=o.provider, cli_bin="lark-cli").owner_attention + path = restarted._path(o.binding["binding_id"]) + state = restarted._read(path) + state["outbox"][0]["next_attempt_at"] = 0 + restarted._write(path, state) + o.provider.verify_replies = True + assert restarted.reconcile() == 1 + assert restarted.reconcile() == 0 and len(o.provider.writes) == before + 1 + assert receipt(o)["status"] == "delivered" and restarted._read(path)["outbox"] == [] + + +@pytest.mark.parametrize("failure", ["timeout", "crash_after_effect", "spawn"]) +def test_write_failure_semantics_survive_process_restart(owner, failure): + o = owner + o.configure() + o.tasks([{"id": "todo_new", "text": "A real new owner task"}]) + original = o.transport.runner + def runner(args, cwd=None, timeout=None): + if "+messages-send" in args: + if failure == "spawn": + raise FileNotFoundError("private credential must not leak") + original(args, cwd, timeout) # effect happened; no locator returned + if failure == "crash_after_effect": + raise KeyboardInterrupt("simulated process crash") + raise subprocess.TimeoutExpired(args, timeout) + return original(args, cwd, timeout) + o.transport.runner = runner + before = len(o.provider.writes) + if failure == "crash_after_effect": + with pytest.raises(KeyboardInterrupt): + o.attention.reconcile() + else: + o.attention.reconcile() + o.transport.runner = original + restarted = LarkPrivateConversations(controller=o.runtime, runtime_root=o.store.root.parent, + runner=o.provider, cli_bin="lark-cli").owner_attention + path = restarted._path(o.binding["binding_id"]) + state = restarted._read(path) + state["outbox"][0]["next_attempt_at"] = 0 + restarted._write(path, state) + restarted.reconcile() + if failure == "spawn": + assert len(o.provider.writes) == before + 1 and receipt(o)["status"] == "delivered" + else: + assert len(o.provider.writes) == before + 1 and restarted._read(path)["outbox"][0]["status"] == "unconfirmed" + + +@pytest.mark.parametrize("denial", ["wrong_app", "wrong_owner", "source_changed", "revoked", "disabled", "source_unavailable"]) +def test_current_audience_scope_and_settings_remain_required(owner, denial): + o = owner + o.configure() + o.tasks([{"id": "todo_new", "text": "A source-owned decision"}]) + if denial == "wrong_app": + o.provider.profile_apps["steward-app"] = "cli_foreign" + elif denial == "wrong_owner": + o.provider.owner_ready = False + elif denial == "source_changed": + o.provider.messages[o.event["message_id"]]["chat_id"] = "oc_foreign" + elif denial == "revoked": + o.transport.bindings.disconnect(o.binding["binding_id"], expected_revision=o.transport.bindings.read()["revision"]) + elif denial == "disabled": + o.configure(False) + else: + o.runtime.registry_path.write_text("not valid JSON") + before = len(o.provider.writes) + o.attention.reconcile() + assert len(o.provider.writes) == before + assert "private credential" not in json.dumps(o.attention.health(o.binding["binding_id"])) + + +def test_withdrawal_disablement_and_new_goal_after_enable(owner): + o = owner + o.configure() + o.tasks([{"id": "todo_new", "text": "New registered Goal needs an owner"}], goal="beta") + o.configure(False) + assert o.attention.reconcile() == 0 + o.configure(True) + path = o.attention._path(o.binding["binding_id"]) + assert len(o.attention._read(path)["outbox"]) == 1 + o.tasks([{"id": "todo_new", "text": "New registered Goal needs an owner", "done": True}], goal="beta") + before = len(o.provider.writes) + assert o.attention.reconcile() == 0 and len(o.provider.writes) == before + assert receipt(o)["status"] == "superseded" + + +def test_real_settings_endpoint_cas_and_disablement_without_provider(owner): + o = owner + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.chat_store, server.runtime_controller, server.verbose = o.store, o.runtime, False + server.lark_private_conversations = o.transport + server.lark_goal_topic_runtime = SimpleNamespace(health_snapshot=lambda: {}, close=lambda: None) + worker = threading.Thread(target=server.serve_forever, daemon=True) + worker.start() + def post(body): + c = http.client.HTTPConnection(*server.server_address, timeout=15) + try: + c.request("POST", "/api/chat/lark/private-conversations/owner-notifications", json.dumps(body), {"Content-Type": "application/json"}) + response = c.getresponse() + return response.status, json.loads(response.read()) + finally: + c.close() + body = {"binding_id": o.binding["binding_id"], "binding_revision": o.transport.bindings.read()["revision"], "revision": 0, "enabled": True} + try: + code, payload = post(body) + assert code == 200 + row = next(row for row in payload["connections"] if row["binding_id"] == o.binding["binding_id"]) + assert row["owner_attention"]["enabled"] and row["owner_attention"]["revision"] == 1 + assert post(body)[0] == 400, "stale settings cannot overwrite a newer choice" + assert post({**body, "revision": 1, "chat_id": "oc_foreign"})[0] == 400 + assert post({**body, "binding_id": o.transport._binding("notes-app")["binding_id"]})[0] == 400 + o.provider.owner_ready = False + assert post({**body, "revision": 1, "enabled": False})[0] == 200, "local disable does not depend on network auth" + assert not o.attention.health(o.binding["binding_id"])["enabled"] + finally: + server.shutdown() + server.server_close() + worker.join(2) + + +@pytest.mark.parametrize("provider_kind", ["file", "sqlite"]) +def test_discovery_uses_promoted_native_authority_not_stale_markdown(owner, tmp_path, monkeypatch, provider_kind): + from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + o = owner + isolate_sqlite_runtime(tmp_path, monkeypatch) + o.configure() + o.tasks([{"id": "todo_stale", "text": "Stale markdown must not notify"}]) + todos = [{"schema_version": "todo_item_v0", "todo_id": "todo_native", "role": "user", "status": "open", + "done": False, "text": "Current canonical owner decision", "task_class": "user_gate", + "archive_state": "active", "source_section": "User Todo", "index": 1}] + projection = build_todo_runtime_shadow_projection(goal_id="alpha", todos=todos, handoff_mode="soft_claim") + initialize_canonical_authority(o.runtime.coordination_runtime_root, "alpha", projection, + state_path=tmp_path / "alpha.md", provider=provider_kind) + before = len(o.provider.writes) + assert o.attention.reconcile() == 1 + assert len(o.provider.writes) == before + 1 + assert "Current canonical owner decision" in o.provider.writes[-1][1] + assert "Stale markdown" not in o.provider.writes[-1][1] + + +@pytest.mark.parametrize("mismatch", ["app", "chat", "text"]) +def test_independent_readback_cannot_borrow_neighbor_identity_or_body(owner, mismatch): + o = owner + o.configure() + o.tasks([{"id": "todo_new", "text": "Verify the exact delivered message"}]) + original = o.transport.runner + def runner(args, cwd=None, timeout=None): + result = original(args, cwd, timeout) + if "+messages-mget" in args and args[args.index("--message-ids") + 1].startswith("om_out_"): + payload = json.loads(result["stdout"]) + message = payload["data"]["items"][0] + if mismatch == "app": + message["sender"] = {"sender_type": "app", "id": "cli_neighbor"} + elif mismatch == "chat": + message["chat_id"] = "oc_neighbor" + else: + message["body"] = {"content": json.dumps({"text": "Only part of the answer"})} + result = {**result, "stdout": json.dumps(payload)} + return result + o.transport.runner = runner + before = len(o.provider.writes) + assert o.attention.reconcile() == 0 + assert o.attention.health(o.binding["binding_id"])["unconfirmed_count"] == 1 + assert len(o.provider.writes) == before + 1 From 2dc596fa6bf8f3fa86d67f3ee3ae248db7dd8156 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:23:22 +0800 Subject: [PATCH 2/4] fix(lark): preserve reply capacity when owner notices are disabled Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/capabilities/manager_context/README.md | 4 ++- .../lark/goal_topic_runtime_service.py | 3 ++- loopx/extensions/lark/owner_attention.py | 12 +++++++++ tests/test_lark_owner_attention.py | 26 +++++++++++++++++++ 4 files changed, 43 insertions(+), 2 deletions(-) diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index 1633c217cd..9e0cdf45c7 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -174,7 +174,9 @@ owns subject identity, meaningful-content changes and reopening generations; this is a local notification journal, not another Todo authority. The existing four-worker outbound pool reserves at most one slot for bounded -notification reconciliation. Each durable per-binding outbox has a process +notification reconciliation only when a notification setting is enabled; +the default-off path retains all four reply slots and performs no Todo scan. +Each durable per-binding outbox has a process fence. Before each send, Core rechecks the exact steward, App, owner, current scope and original private-message provenance. Settings cannot supply another destination. A frozen intent precedes the provider write; the independent diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index e4dbef48f0..93b912dd0c 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -106,7 +106,8 @@ def _reconcile_private(self) -> None: paths = transport.pending_delivery_paths() attempted = {path: count for path, count in attempted.items() if path in paths} owner_attention = getattr(transport, "owner_attention", None) - attention_ready = owner_attention is not None and attention is None and time.monotonic() >= attention_due + attention_ready = (owner_attention is not None and attention is None + and time.monotonic() >= attention_due and owner_attention.has_enabled()) for path in sorted(paths, key=lambda path: attempted.get(path, 0)): if len(pending) + int(attention is not None or attention_ready) >= 4 or self._closed.is_set(): break diff --git a/loopx/extensions/lark/owner_attention.py b/loopx/extensions/lark/owner_attention.py index 50654cd71e..eb9610f766 100644 --- a/loopx/extensions/lark/owner_attention.py +++ b/loopx/extensions/lark/owner_attention.py @@ -69,6 +69,18 @@ def health(self, binding_id: str) -> dict[str, Any]: except (ValueError, OSError, KeyError, TypeError, RuntimeError): return {"available": False, "error_code": "notification_state_unavailable"} + def has_enabled(self) -> bool: + """A local setting alone activates work; installation never reserves IO.""" + for path in self.root.glob("*.json"): + if not re.fullmatch(r"[a-f0-9]{24}", path.stem): + continue + try: + if self._read(path).get("enabled") is True: + return True + except (ValueError, OSError, KeyError, TypeError): + continue + return False + def _binding(self, binding_id: str, expected_revision: int | None = None) -> dict[str, Any]: current = self.transport.bindings.read() if expected_revision is not None and current["revision"] != expected_revision: diff --git a/tests/test_lark_owner_attention.py b/tests/test_lark_owner_attention.py index 2ca4029aea..693140aa7c 100644 --- a/tests/test_lark_owner_attention.py +++ b/tests/test_lark_owner_attention.py @@ -5,6 +5,7 @@ import threading import time from types import SimpleNamespace +from pathlib import Path import pytest @@ -82,6 +83,31 @@ def test_service_discovers_new_native_tasks_without_model_or_agent_poll(owner): worker.join(10) +def test_disabled_notifications_preserve_all_four_existing_delivery_slots(owner, monkeypatch): + o = owner + entered, release = [], threading.Event() + def existing_reply(path): + entered.append(path) + release.wait(5) + return 0 + monkeypatch.setattr(o.transport, "pending_delivery_paths", lambda: [Path(str(i)) for i in range(4)]) + monkeypatch.setattr(o.transport, "reconcile_request", existing_reply) + monkeypatch.setattr(o.attention, "reconcile", lambda: pytest.fail("disabled notifications must not schedule a job")) + service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=o.store.root.parent, + runtime_controller=o.runtime, private_conversations=o.transport) + worker = threading.Thread(target=service._reconcile_private, daemon=True) + worker.start() + try: + deadline = time.monotonic() + 3 + while len(entered) < 4 and time.monotonic() < deadline: + time.sleep(.01) + assert len(entered) == 4, "disabled feature retains the pre-change outbound concurrency" + finally: + service._closed.set() + release.set() + worker.join(5) + + def test_restart_verifies_original_locator_and_never_resends(owner): o = owner o.configure() From 0b4e912edba297b0175838105fb0c5026b571c90 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:28:13 +0800 Subject: [PATCH 3/4] test(lark): hold the scan boundary in disabled-capacity regression Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_lark_owner_attention.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/test_lark_owner_attention.py b/tests/test_lark_owner_attention.py index 693140aa7c..d0a288379a 100644 --- a/tests/test_lark_owner_attention.py +++ b/tests/test_lark_owner_attention.py @@ -85,9 +85,11 @@ def test_service_discovers_new_native_tasks_without_model_or_agent_poll(owner): def test_disabled_notifications_preserve_all_four_existing_delivery_slots(owner, monkeypatch): o = owner - entered, release = [], threading.Event() + entered, release, all_started = [], threading.Event(), threading.Event() def existing_reply(path): entered.append(path) + if len(entered) == 4: + all_started.set() release.wait(5) return 0 monkeypatch.setattr(o.transport, "pending_delivery_paths", lambda: [Path(str(i)) for i in range(4)]) @@ -95,12 +97,13 @@ def existing_reply(path): monkeypatch.setattr(o.attention, "reconcile", lambda: pytest.fail("disabled notifications must not schedule a job")) service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=o.store.root.parent, runtime_controller=o.runtime, private_conversations=o.transport) + # Hold the scan boundary: an empty optional job cannot borrow a reply + # slot and defer the fourth existing request until a later scan. + monkeypatch.setattr(service._closed, "wait", lambda _: release.wait(5)) worker = threading.Thread(target=service._reconcile_private, daemon=True) worker.start() try: - deadline = time.monotonic() + 3 - while len(entered) < 4 and time.monotonic() < deadline: - time.sleep(.01) + assert all_started.wait(2) assert len(entered) == 4, "disabled feature retains the pre-change outbound concurrency" finally: service._closed.set() From 5e369c3e0f2ea9d333674d6ea6f1f3860ef25c97 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:38:05 +0800 Subject: [PATCH 4/4] refactor(lark): reuse attention eligibility and skip terminal body reads Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../control_plane/effect_runtime_handlers.ts | 1 + .../presentation/owner_attention.ts | 19 +++++++++++++++---- loopx/extensions/lark/owner_attention.py | 10 ++++++---- .../control_plane_ts/owner_attention.test.ts | 8 +++++++- 4 files changed, 29 insertions(+), 9 deletions(-) diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index cb0af7c041..dd21cf752f 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -460,6 +460,7 @@ export function createEffectRuntimeHandlers( ["presentation.owner_attention.plan", lazyHandler(() => import("./presentation/owner_attention.ts"), ({planOwnerAttention}) => planOwnerAttention)], ["presentation.owner_attention.observe", lazyHandler(() => import("./presentation/owner_attention.ts"), ({observeOwnerAttention}) => observeOwnerAttention)], ["presentation.owner_attention.health", lazyHandler(() => import("./presentation/owner_attention.ts"), ({ownerAttentionHealth}) => ownerAttentionHealth)], + ["presentation.owner_attention.references", lazyHandler(() => import("./presentation/owner_attention.ts"), ({ownerAttentionReferences}) => ownerAttentionReferences)], ["presentation.goal_attention.bound", lazyHandler(() => import("./presentation/goal_attention.ts"), ({boundGoalAttention}) => boundGoalAttention)], ["presentation.goal_attention.reconcile", lazyHandler(() => import("./presentation/goal_attention.ts"), ({reconcileGoalAttention}) => reconcileGoalAttention)], ["presentation.action_review_plan.compile", lazyHandler(() => import("./presentation/action_review_plan.ts"), ({compileActionReviewPlan}) => (params) => diff --git a/loopx/control_plane/presentation/owner_attention.ts b/loopx/control_plane/presentation/owner_attention.ts index fd788cf6f7..8697e6a910 100644 --- a/loopx/control_plane/presentation/owner_attention.ts +++ b/loopx/control_plane/presentation/owner_attention.ts @@ -3,14 +3,25 @@ import type {JsonObject} from "../effect_program.ts"; import {requireInteger, requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; import {canonicalAuthoritySha256 as digest} from "../coordination/authority_store_codec.ts"; +import {projectGoalAttention} from "./goal_attention.ts"; const records = (value: unknown, name: string): JsonObject[] => { if (!Array.isArray(value)) throw new TypeError(`${name} must be a complete array`); return value.map(row => requireJsonObject(row, name)); }; -const active = (row: JsonObject) => row.role === "user" && row.done !== true - && !row.superseded_by && ["open", "blocked"].includes(String(row.status)) - && ["user_action", "user_gate"].includes(String(row.task_class)); +function active(row: JsonObject, goalId: string): boolean { + if (row.goal_id != null && row.goal_id !== goalId) throw new Error("foreign owner Todo"); + // Reuse the existing attention owner, one source at a time. Its presentation + // bound cannot omit this single subject; incomplete notice terms are still + // restored independently from the native exact reader before delivery. + return projectGoalAttention({goal_id: goalId, todos: [row], blockers: [], limit: 1}).items.length === 1; +} + +export function ownerAttentionReferences(input: JsonObject): JsonObject { + const goalId = requireNonEmptyString(input.goal_id, "goal_id"); + return {todo_ids: records(input.todos, "todos").filter(row => active(row, goalId)) + .map(row => requireNonEmptyString(row.todo_id, "todo_id"))}; +} /** Local provider-journal vocabulary; never canonical Todo/Run status. */ function readState(raw: unknown): JsonObject { @@ -45,7 +56,7 @@ export function planOwnerAttention(input: JsonObject): JsonObject { for (const goal of records(input.goals, "goals")) { const goalId = requireNonEmptyString(goal.goal_id, "goal_id"); for (const todo of records(goal.todos, "todos")) { - if (!active(todo)) continue; + if (!active(todo, goalId)) continue; if (todo.goal_id != null && todo.goal_id !== goalId) throw new Error("foreign owner Todo"); const todoId = requireNonEmptyString(todo.todo_id, "todo_id"); const text = requireNonEmptyString(todo.text, "todo.text"); diff --git a/loopx/extensions/lark/owner_attention.py b/loopx/extensions/lark/owner_attention.py index eb9610f766..618bb9a36e 100644 --- a/loopx/extensions/lark/owner_attention.py +++ b/loopx/extensions/lark/owner_attention.py @@ -134,10 +134,12 @@ def _snapshot(self, binding: dict[str, Any]) -> list[dict[str, Any]]: if listing.get("ok") is not True or listing.get("todo_count") != len(listing.get("todos", [])): raise ValueError("the native Todo source is incomplete") todos = [] - # Exact reads restore source text; a bounded summary never supplies - # notification terms. Read every listed user row without a prefix. - for row in listing["todos"]: - detail = list_goal_todos(**args, todo_id=row["todo_id"]) + # The same typed owner selects the complete active population. + # Terminal history needs no full-body read on every service scan; + # bounded summaries never supply the actual notification terms. + refs = effect_runtime_result("presentation.owner_attention.references", {"goal_id": goal_id, "todos": listing["todos"]}) + for todo_id in refs["todo_ids"]: + detail = list_goal_todos(**args, todo_id=todo_id) if detail.get("ok") is not True or not detail.get("matched") or not isinstance(detail.get("todo"), dict): raise ValueError("a native Todo changed while reading") todos.append(detail["todo"]) diff --git a/tests/control_plane_ts/owner_attention.test.ts b/tests/control_plane_ts/owner_attention.test.ts index bd2841a01c..98001f553e 100644 --- a/tests/control_plane_ts/owner_attention.test.ts +++ b/tests/control_plane_ts/owner_attention.test.ts @@ -1,6 +1,6 @@ import assert from "node:assert/strict"; import test from "node:test"; -import {observeOwnerAttention, ownerAttentionHealth, planOwnerAttention} from "../../loopx/control_plane/presentation/owner_attention.ts"; +import {observeOwnerAttention, ownerAttentionHealth, ownerAttentionReferences, planOwnerAttention} from "../../loopx/control_plane/presentation/owner_attention.ts"; import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; const empty = () => ({schema_version: "lark_owner_attention_v0", revision: 0, enabled: true, initialized: false, subjects: [], outbox: []}); @@ -69,3 +69,9 @@ test("unknown journal states cannot present healthy or authorize delivery", () = assert.throws(() => plan(broken, [task])); assert.throws(() => ownerAttentionHealth({state: {...empty(), enabled: "true"}})); }); + +test("reference discovery and planning share active-subject semantics without a display cap", () => { + const tasks = Array.from({length: 80}, (_, i) => ({...task, todo_id: `todo_${i}`})); + const todos = [...tasks, {...task, todo_id: "todo_done", done: true}, {...task, role: "agent"}]; + assert.deepEqual(ownerAttentionReferences({goal_id: "alpha", todos}).todo_ids, tasks.map(row => row.todo_id)); +});