feat: show whether bound Codex threads are running in the App (#5081)

* feat(status): observe bound Codex thread activity for the App status route

Add a read-only host thread observer contract with typed states
(turn_open, idle, archived, unknown with a typed reason) and a Codex
adapter that reads the local thread store and rollout turn markers.
Unrecognized store or record shapes are unknown, never an open turn.
Only the App status route attaches the result; other collect_status
callers are unchanged.

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

* feat(app): show observed host thread turns as execution

An open host-recorded turn with a recent event runs the Goal with the
live ring; silent turns disclose their silence and turns without an
event for six hours are not execution. Observed idle threads read as
idle instead of deferring to the host.

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(status): keep partial and malformed host observations honest

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>

* fix(status): preserve claim clocks and escaped event semantics

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>

---------

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Co-authored-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: huangruiteng

apps/presentation/dashboard/src/data/status.ts

@@ -527,6 +527,20 @@ export const runGoalSchema = z.object({
       host_surface: z.string().optional().nullable(),
     })).optional().default([]).catch([]),
   }).optional().nullable(),
+  host_thread_activity: z.object({
+    observed_at: z.string().optional().nullable(),
+    // Older or malformed observations cannot prove that every binding was read.
+    completeness: z.enum(["complete", "incomplete"]).catch("incomplete").default("incomplete"),
+    threads: z.array(z.object({
+      agent_id: z.string().optional().nullable(),
+      host_surface: z.string(),
+      state: z.enum(["turn_open", "idle", "archived", "unknown"]).catch("unknown"),
+      reason: z.string().optional().nullable(),
+      turn_started_at: z.string().optional().nullable(),
+      last_turn_ended_at: z.string().optional().nullable(),
+      last_event_at: z.string().optional().nullable(),
+    })).optional().default([]),
+  }).optional().nullable().catch(null),
   index_exists: z.boolean().optional().default(false),
   raw_index_records: z.number().optional().default(0),
   unique_runs: z.number().optional().default(0),

apps/presentation/dashboard/src/features/personal-workspace/goal-activity-view.tsx

@@ -3,7 +3,7 @@ import { goalHostSurfaces, goalIdentity, hostSurfaceLabel, presentGoalActivity,
 import { useWorkspaceI18n, type WorkspaceMessageKey } from "./i18n";
 import type { WorkspaceGoal } from "./personal-workspace-model";
 
-type GoalActivitySubject = Pick<WorkspaceGoal, "activationState" | "boundHostSurfaces" | "execution" | "goalId" | "loadState" | "needsYou" | "state" | "title">;
+type GoalActivitySubject = Pick<WorkspaceGoal, "activationState" | "boundHostSurfaces" | "execution" | "goalId" | "hostThreadActivity" | "loadState" | "needsYou" | "state" | "title">;
 
 export function relativeTime(value: string, locale: string): string | null {
   const then = new Date(value).getTime();

apps/presentation/dashboard/src/features/personal-workspace/goal-activity.test.mjs

@@ -1,5 +1,5 @@
 import assert from "node:assert/strict";
-import { goalExecutionFromSessions, presentGoalActivity } from "../../../node_modules/.cache/loopx-goal-activity/goal-activity.js";
+import { goalExecution, presentGoalActivity } from "../../../node_modules/.cache/loopx-goal-activity/goal-activity.js";
 
 // One Goal's session facts, mixed by mode. `updated_at` stands for the time the
 // owner last recorded anything for that session.
@@ -19,8 +19,8 @@ const recent = minutesAgo(2);
 const silent = minutesAgo(30);
 const old = minutesAgo(45);
 
-function executionOf(sessions) {
-  const execution = goalExecutionFromSessions(sessions, "goal-a", now);
+function executionOf(sessions, observation) {
+  const execution = goalExecution(sessions, "goal-a", observation, now);
   assert.equal(execution.kind, "running");
   return execution;
 }
@@ -66,9 +66,35 @@ assert.equal(presentGoalActivity({ activationState: "active", execution: recentW
 
 // Several managed turns report the newest one; a closed session is not open work.
 assert.equal(executionOf([session("a", "managed", silent), session("b", "managed", recent)]).lastActivityAt, recent);
-assert.equal(goalExecutionFromSessions([{ ...session("closed", "managed", recent), status: "closed" }], "goal-a", now).kind, "idle");
-assert.equal(goalExecutionFromSessions([session("other", "managed", recent, null)], "goal-a", now).kind, "idle");
-assert.equal(goalExecutionFromSessions(null, "goal-a", now).kind, "unknown");
-assert.equal(goalExecutionFromSessions([session("other-goal", "managed", recent)], "goal-b", now).kind, "idle");
+assert.equal(goalExecution([{ ...session("closed", "managed", recent), status: "closed" }], "goal-a", undefined, now).kind, "idle");
+assert.equal(goalExecution([session("other", "managed", recent, null)], "goal-a", undefined, now).kind, "idle");
+assert.equal(goalExecution(null, "goal-a", undefined, now).kind, "unknown");
+assert.equal(goalExecution([session("other-goal", "managed", recent)], "goal-b", undefined, now).kind, "idle");
+
+// Native observation joins actual activity; it does not turn a fresh attached
+// claim into execution or let that claim refresh a silent host Turn.
+const hostObservation = (lastEventAt, completeness = "complete") => ({
+  completeness,
+  threads: [{ hostSurface: "codex-app", state: "turn_open", lastEventAt }],
+});
+const hostWithClaim = executionOf([session("attached", "attached_host", recent)], hostObservation(silent));
+assert.equal(hostWithClaim.lastActivityAt, silent);
+assert.equal(hostWithClaim.claimedAt, recent);
+assert.equal(hostWithClaim.hostClaimed, false);
+assert.equal(hostWithClaim.quiet, true);
+assert.equal(presentGoalActivity({ activationState: "active", execution: hostWithClaim, state: "推进" }).live, false);
+const freshHost = executionOf([session("managed", "managed", silent)], hostObservation(recent, "incomplete"));
+assert.equal(freshHost.lastActivityAt, recent);
+assert.equal(freshHost.quiet, false);
+assert.equal(presentGoalActivity({ activationState: "active", execution: freshHost, state: "推进" }).live, true);
+assert.equal(goalExecution(undefined, "goal-a", hostObservation(recent), now).kind, "running");
+assert.equal(goalExecution(null, "goal-a", hostObservation(minutesAgo(8 * 60)), now).kind, "unknown");
+for (const completeness of ["incomplete", undefined]) {
+  const partialIdle = presentGoalActivity({
+    activationState: "active", state: "已安排", boundHostSurfaces: ["codex-app"],
+    hostThreadActivity: { completeness, threads: [{ hostSurface: "codex-app", state: "idle", lastEventAt: recent }] },
+  });
+  assert.equal(partialIdle.alsoKey, "activity.inHost", "an incomplete sample never proves all threads idle");
+}
 
 console.log("Goal execution read-model invariants passed");

apps/presentation/dashboard/src/features/personal-workspace/goal-activity.ts

@@ -1,25 +1,9 @@
-// Execution is a fact read from the chat/session owner. Open Todos, quota
-// eligibility, registration or a persistent session never imply it.
-// Attached hosts only surface turns they claimed from LoopX, and the claim
-// timestamp is their only activity fact; work a host starts on its own, and
-// every turn of a bound host thread, is invisible here. Such Goals are
-// labelled as host-owned instead of running.
+// Execution is a fact read from the chat/session owner or from a host's own
+// thread records. Open Todos, quota eligibility, registration, a persistent
+// session or a thread binding never imply it. Attached hosts only surface turns
+// they claimed from LoopX, and the claim timestamp is their only activity fact.
 export type WorkspaceGoalExecution =
-  | {
-    kind: "running";
-    agentIds: string[];
-    hostClaimed: boolean;
-    hostSurfaces: string[];
-    /**
-     * Newest activity of a managed turn. An attached claim never writes this:
-     * taking a turn is not evidence that anything executed.
-     */
-    lastActivityAt: string | null;
-    /** Newest claim time of an attached turn, reported as a claim rather than as activity. */
-    claimedAt: string | null;
-    quiet: boolean;
-    sessionCount: number;
-  }
+  | { kind: "running"; hostClaimed: boolean; hostSurfaces: string[]; lastActivityAt: string | null; claimedAt: string | null; quiet: boolean }
   | { kind: "idle"; hostSurfaces: string[] }
   | { kind: "unknown" };
 
@@ -34,8 +18,23 @@ export type GoalSessionFact = {
   updated_at?: string | null;
 };
 
-/** A claimed turn with no recorded event for this long may be silent or lost; it is never shown as live. */
+/** One bound host thread as observed by the host adapter (`host_thread_activity`). */
+export type GoalHostThread = {
+  hostSurface: string;
+  state: "turn_open" | "idle" | "archived" | "unknown";
+  lastEventAt: string | null;
+};
+
+/** Coverage stays with the rows so a bounded sample cannot prove all hosts idle. */
+export type GoalHostThreadActivity = {
+  completeness: "complete" | "incomplete";
+  threads: GoalHostThread[];
+};
+
+/** A turn with no recorded event for this long may be silent or lost; it is never shown as live. */
 export const quietTurnMinutes = 15;
+/** A host can exit mid-turn without recording an end; after this long an open turn is not execution. */
+export const abandonedHostTurnHours = 6;
 
 function hostSurfacesOf(sessions: readonly GoalSessionFact[]) {
   return Array.from(new Set(sessions
@@ -56,30 +55,47 @@ function newestRecordedTime(sessions: readonly GoalSessionFact[]) {
     .at(-1) ?? null;
 }
 
-/** `sessions === null` means the session owner could not be read. */
-export function goalExecutionFromSessions(sessions: readonly GoalSessionFact[] | null, goalId: string, now = Date.now()): WorkspaceGoalExecution {
-  if (sessions === null) return { kind: "unknown" };
-  const open = sessions.filter((session) => session.goal_id === goalId && session.status !== "closed");
+function openHostTurns(threads: readonly GoalHostThread[], now: number) {
+  return threads.filter((thread) => {
+    const lastEventMs = thread.lastEventAt ? Date.parse(thread.lastEventAt) : Number.NaN;
+    return thread.state === "turn_open" && !Number.isNaN(lastEventMs) && now - lastEventMs <= abandonedHostTurnHours * 3_600_000;
+  });
+}
+
+/**
+ * `sessions === null` means the session owner could not be read; `undefined`
+ * means it has not been read yet, so only host-observed turns are known.
+ */
+export function goalExecution(
+  sessions: readonly GoalSessionFact[] | null | undefined,
+  goalId: string,
+  hostThreadActivity?: GoalHostThreadActivity,
+  now = Date.now(),
+): WorkspaceGoalExecution | undefined {
+  const hostTurns = openHostTurns(hostThreadActivity?.threads ?? [], now);
+  const open = (sessions ?? []).filter((session) => session.goal_id === goalId && session.status !== "closed");
   const active = open.filter((session) => Boolean(session.active_turn_id));
-  if (active.length === 0) return { kind: "idle", hostSurfaces: hostSurfacesOf(open) };
-  // Aggregate by session mode: a managed turn carries execution activity, an
-  // attached turn carries only its claim. Reading one maximum across both let a
-  // recent claim make a silent managed turn look live.
+  if (active.length === 0 && hostTurns.length === 0) {
+    if (sessions === undefined) return undefined;
+    return sessions === null ? { kind: "unknown" } : { kind: "idle", hostSurfaces: hostSurfacesOf(open) };
+  }
+  // Claims retain their own clock. Only managed activity and observed host
+  // events can refresh execution, including in mixed-source observations.
   const managedTurns = active.filter((session) => !isAttachedTurn(session));
-  const attachedTurns = active.filter(isAttachedTurn);
-  const lastActivityAt = newestRecordedTime(managedTurns);
-  const claimedAt = newestRecordedTime(attachedTurns);
-  const hostClaimed = managedTurns.length === 0;
+  const claimedAt = newestRecordedTime(active.filter(isAttachedTurn));
+  const lastActivityAt = [
+    newestRecordedTime(managedTurns),
+    ...hostTurns.map((thread) => thread.lastEventAt),
+  ].filter(Boolean).sort().at(-1) ?? null;
   const lastActivityMs = lastActivityAt ? Date.parse(lastActivityAt) : Number.NaN;
+  const hostClaimed = hostTurns.length === 0 && managedTurns.length === 0;
   return {
     kind: "running",
-    agentIds: Array.from(new Set(active.map((session) => session.agent_id))),
     hostClaimed,
-    hostSurfaces: hostSurfacesOf(active),
+    hostSurfaces: Array.from(new Set([...hostSurfacesOf(active), ...hostTurns.map((thread) => thread.hostSurface)])),
     lastActivityAt,
     claimedAt,
-    quiet: !Number.isNaN(lastActivityMs) && now - lastActivityMs > quietTurnMinutes * 60_000,
-    sessionCount: active.length,
+    quiet: !hostClaimed && !Number.isNaN(lastActivityMs) && now - lastActivityMs > quietTurnMinutes * 60_000,
   };
 }
 
@@ -88,6 +104,8 @@ const hostSurfaceNames: Readonly<Record<string, string>> = {
   "claude-code": "Claude Code",
   "codex-app": "Codex App",
   "codex-app-ssh": "Codex App (SSH)",
+  "codex-cli-tui": "Codex CLI",
+  "codex-ide-plugin": "Codex IDE",
   cursor: "Cursor",
   kiro: "Kiro",
 };
@@ -119,6 +137,7 @@ type GoalActivityInput = {
   /** `host_surface` of each thread bound to this Goal in the status projection. */
   boundHostSurfaces?: string[];
   execution?: WorkspaceGoalExecution;
+  hostThreadActivity?: GoalHostThreadActivity;
   needsYou?: string | null;
   state: string;
 };
@@ -134,7 +153,12 @@ export function presentGoalActivity(goal: GoalActivityInput): GoalActivity {
   if (running) return { labelKey: "activity.running", tone: live ? "running" : "attention", live, alsoKey: null };
   if (goal.state === "等待条件") return { labelKey: "state.waiting", tone: "waiting", live: false, alsoKey: null };
   if (goal.state === "已安排") {
-    const alsoKey = goal.execution?.kind === "unknown" ? "activity.executionUnknown" : goalHostSurfaces(goal).length > 0 ? "activity.inHost" : null;
+    const hostThreads = goal.hostThreadActivity?.threads ?? [];
+    const hostsIdle = goal.hostThreadActivity?.completeness === "complete"
+      && hostThreads.length > 0 && hostThreads.every((thread) => thread.state === "idle" || thread.state === "archived");
+    const alsoKey = goal.execution?.kind === "unknown" ? "activity.executionUnknown"
+      : hostsIdle ? "activity.hostIdle"
+        : goalHostSurfaces(goal).length > 0 ? "activity.inHost" : null;
     return { labelKey: "state.queued", tone: "queued", live: false, alsoKey };
   }
   if (goal.state === "已完成") return { labelKey: "state.completed", tone: "quiet", live: false, alsoKey: null };

apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx

@@ -1055,6 +1055,7 @@ const en = {
   "activity.executionUnknown": "run state unavailable",
   "activity.inHost": "handled by a {host} thread; the host owns live state",
   "activity.hostClaimed": "Claimed by host",
+  "activity.hostIdle": "{host} thread idle",
   "activity.claimedAt": "claimed {time}",
   "activity.viaHost": "{host} host",
   "activity.quiet": "no new activity for {minutes} min",
@@ -2201,6 +2202,7 @@ const zhCN: Record<WorkspaceMessageKey, string> = {
   "activity.executionUnknown": "暂时读不到执行状态",
   "activity.inHost": "由 {host} 线程负责，执行情况以宿主为准",
   "activity.hostClaimed": "宿主已领取",
+  "activity.hostIdle": "{host} 线程空闲",
   "activity.claimedAt": "领取于{time}",
   "activity.viaHost": "{host} 宿主",
   "activity.quiet": "已 {minutes} 分钟无新动静",

apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts

@@ -4,7 +4,7 @@ import type { ActionReviewPlan } from "../../../../../../loopx/control_plane/pre
 import type { GoalAcceptanceObservation } from "../../data/goal-acceptance-observation";
 import type { AttentionDetails } from "./attention-details";
 import type { WorkspaceLoadError } from "../../data/workspace-progressive-status";
-import type { WorkspaceGoalExecution } from "./goal-activity";
+import type { GoalHostThreadActivity, WorkspaceGoalExecution } from "./goal-activity";
 export type WorkspaceGoalState =
   | "需修复"
   | "等你"
@@ -107,6 +107,8 @@ export type WorkspaceGoal = {
   doneTodoCount?: number;
   execution?: WorkspaceGoalExecution;
   goalId: string;
+  /** Bound host threads as observed from the host's own records. */
+  hostThreadActivity?: GoalHostThreadActivity;
   latestActivity?: string;
   needsYou?: string | null;
   needsYouBlocking?: boolean;

apps/presentation/dashboard/src/views/dashboard-page.tsx

@@ -113,7 +113,7 @@ import {
   type WorkspaceActionPreview,
   type WorkspaceActionPreviewRequest,
 } from "../features/personal-workspace/personal-workspace-model";
-import { goalExecutionFromSessions } from "../features/personal-workspace/goal-activity";
+import { goalExecution } from "../features/personal-workspace/goal-activity";
 
 const protectedOperationLabels: Record<ProtectedActionProposal["operation"], string> = {
   delete: "删除",
@@ -1112,6 +1112,14 @@ function buildPersonalHomeModel(
       boundHostSurfaces: Array.from(new Set((goal.coordination?.thread_agent_bindings ?? [])
         .flatMap((binding) => binding.host_surface ? [binding.host_surface] : []))),
       doneTodoCount: agentTodoFacts.doneTodoCount,
+      hostThreadActivity: goal.host_thread_activity ? {
+        completeness: goal.host_thread_activity.completeness,
+        threads: goal.host_thread_activity.threads.map((thread) => ({
+          hostSurface: thread.host_surface,
+          lastEventAt: thread.last_event_at ?? null,
+          state: thread.state,
+        })),
+      } : undefined,
       acceptanceObservation: goal.acceptance_observation,
       goalId: goal.id,
       latestActivity: row.latestRun?.generated_at ?? "",
@@ -2627,9 +2635,9 @@ function PersonalGoalHome({
   const normalizedModel = normalizePersonalHomeModel(model);
   const workspaceModel = {
     ...normalizedModel,
-    goals: goalSessionFacts === undefined ? normalizedModel.goals : normalizedModel.goals.map((goal) => ({
+    goals: normalizedModel.goals.map((goal) => ({
       ...goal,
-      execution: goalExecutionFromSessions(goalSessionFacts, goal.goalId),
+      execution: goalExecution(goalSessionFacts, goal.goalId, goal.hostThreadActivity),
     })),
     userTodos: model.userTodos.map(attentionForWorkspace),
     attentionHistory: (model.attentionHistory ?? model.userTodos).map(attentionForWorkspace),

docs/integrations/attached-agent-session-broker.md

@@ -72,3 +72,36 @@ Opaque host identifiers, message bodies, and response files stay in the local
 runtime store. Public Session projections contain only the LoopX Session id,
 Goal/Agent binding, executor endpoint label, host surface, capability booleans,
 and lifecycle state.
+
+## Bound Codex Thread Activity
+
+The broker only sees turns a host claims. A host thread registered with
+`bind-agent-thread` but never attached is observed separately, read-only, from
+the host's own local store. The App status route (`/status.json`) adds
+`run_history.goals[].host_thread_activity` with up to 32 bound-thread rows:
+`agent_id`, `host_surface`, `state`, and the `turn_started_at`,
+`last_turn_ended_at` and `last_event_at` timestamps when known. Thread ids,
+paths and message content are not included.
+
+The observation's `completeness` is `complete` only when every valid binding
+fits within that read budget; otherwise it is `incomplete`. The App can show
+an observed recent open turn from either, but reports all hosts idle only from
+a complete observation containing only idle or archived threads. Missing or
+unrecognized completeness is treated as incomplete. Malformed event payloads
+make that thread `unknown` without failing the status route; a later recognized
+turn marker restores its observed state.
+
+| `state` | Meaning |
+| --- | --- |
+| `turn_open` | The host recorded a turn start without an end. A host that exits mid-turn leaves this behind, so the App shows it as running only while `last_event_at` is recent. |
+| `idle` | The latest recorded turn ended. |
+| `archived` | The host archived the thread. |
+| `unknown` | Not observable; `reason` is `unsupported_host`, `store_unavailable`, `thread_not_found`, `record_unrecognized` or `no_turn_marker`. |
+
+Codex local surfaces (`codex-app`, `codex-cli-tui`, `codex-ide-plugin`) are read
+from `<CODEX_HOME>/state_<n>.sqlite` and the thread's rollout JSONL. The Codex
+store is not a public contract: any shape the adapter does not recognize is
+`unknown`, never `turn_open`. Remote surfaces such as `codex-app-ssh` are
+`unsupported_host`. Homes are searched in the order `CODEX_HOME`, `~/.codex`,
+then sibling `~/.codex-*` directories; set `LOOPX_CODEX_HOMES` (an
+`os.pathsep`-separated list) to search exactly those homes instead.

examples/personal-workspace-browser/fixture.mjs

@@ -453,6 +453,7 @@ export async function installApi(page, { goalSubagentConfigurationEnabled = true
     },
     operatorCredentialWrites: [],
     turnRequests: [],
+    hostThreadActivity: {},
     answerForMessage: null,
     loopxModeRequests: [],
     get larkConnections() { return runtime.larkConnections; },
@@ -687,6 +688,10 @@ export async function installApi(page, { goalSubagentConfigurationEnabled = true
         },
       );
     }
+    for (const [goalId, activity] of Object.entries(state.hostThreadActivity)) {
+      const goal = fixture.run_history.goals.find((item) => item.id === goalId);
+      if (goal) goal.host_thread_activity = activity;
+    }
     const goalActivationScope = new URL(route.request().url()).searchParams.get("goal_activation");
     const isActiveScope = goalActivationScope === "active";
     const activeGoalCount = fixture.run_history.goals.filter((goal) => goal.activation_state !== "stopped").length;