Merge pull request #5368 from loopx-project/codex/app-owner-controls-1001

Author: huangruiteng

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

@@ -703,6 +703,11 @@ export function chatSessionQueuesFollowUps(session: Pick<ChatSessionSummary, "se
   return session.session_mode === "attached_host";
 }
 
+/** Native steering is offered only by the managed Codex adapter; attached follow-ups keep their queue contract. */
+export function chatSessionSupportsSteering(session: Pick<ChatSessionSummary, "session_mode" | "adapter_kind">) {
+  return session.session_mode !== "attached_host" && session.adapter_kind === "codex_app_server";
+}
+
 export type ManagerRuntimeSessionReadback = {
   schema_version: "manager_runtime_session_readback_v0";
   runtime_profile: "restricted" | "trusted_owner";

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

@@ -750,6 +750,7 @@ function readImageAttachment(file: File, t: WorkspaceTranslate): Promise<Workspa
 
 export function PersonalWorkspacePage({
   conversationQueuesFollowUps = false,
+  conversationSupportsSteering = false,
   conversationSessionId,
   conversationHistoryState,
   agents = [{ agentId: "codex", available: true, capability: "代码与项目执行", label: "Codex" }],
@@ -767,6 +768,8 @@ export function PersonalWorkspacePage({
 }: {
   /** The bound Session's mode queues a message sent while its Turn runs. */
   conversationQueuesFollowUps?: boolean;
+  /** The bound managed executor offers native exact-turn steering. */
+  conversationSupportsSteering?: boolean;
   conversationSessionId?: string;
   conversationHistoryState?: ConversationHistoryStatus;
   agents?: WorkspaceAgentOption[];
@@ -806,6 +809,8 @@ export function PersonalWorkspacePage({
     }
   });
   const [sending, setSending] = useState(false);
+  const [steering, setSteering] = useState(false);
+  const steeringRequests = useRef(new Map<string, { sessionId: string; turnId: string; text: string; id: string }>());
   const [actionDraft, setActionDraft] = useState<WorkspaceActionDraft | null>(null);
   const [loopxMode, setLoopxMode] = useState<LoopXModeSnapshot | null>(null);
   const [loopxDelivery, setLoopxDelivery] = useState<"queue" | "inbox" | "steer">("queue");
@@ -848,8 +853,9 @@ export function PersonalWorkspacePage({
     input.style.height = "auto";
     input.style.height = `${Math.min(input.scrollHeight, 120)}px`;
   }, [composer, selectedGoalId, managerChatOpen]);
-  function setComposerDraft(key: string, value: string) {
+  function setComposerDraft(key: string, value: string, expectedValue?: string) {
     setDrafts((current) => {
+      if (expectedValue !== undefined && current[key] !== expectedValue) return current;
       const next = { ...current };
       if (value) {
         next[key] = value;
@@ -1027,18 +1033,17 @@ export function PersonalWorkspacePage({
       setGoalConversationReceiptVisible(true);
     }
   }, [goalMessages, selectedGoal, selectedGoalTab]);
-  // A managed runtime Session admits one Turn at a time. While the current
-  // Session shows a Turn in flight, a new message would only be rejected, so
-  // the composer waits and points to the reply's own adjust/interrupt
-  // controls. Two deliveries stay open because the service queues them behind
-  // the running Turn: LoopX mode through its own queue, and any message to an
-  // attached host Session.
+  // One composer for both conversations. Running managed Codex work receives
+  // exact-turn instructions; attached hosts and LoopX mode keep their queues.
   const loopxDeliveryOpen = Boolean(conversationSessionId && loopxMode?.session_id === conversationSessionId
     && loopxMode?.enabled && loopxMode.active_turn_id);
-  const conversationTurnRunning = !loopxDeliveryOpen && !conversationQueuesFollowUps && Boolean(conversationSessionId)
-    && managerMessages.some((message) => message.pending && Boolean(message.sourceTurnId)
-      && message.sourceSessionId === conversationSessionId);
-  const composerBlocked = sending || conversationTurnRunning;
+  const runningMessage = managerMessages.find((message) => message.pending && Boolean(message.sourceTurnId)
+    && message.sourceSessionId === conversationSessionId);
+  const conversationTurnRunning = !loopxDeliveryOpen && !conversationQueuesFollowUps && Boolean(runningMessage);
+  const steeringTurnId = conversationTurnRunning && conversationSupportsSteering && !readOnly
+    && callbacks.onSteerConversationTurn ? runningMessage?.sourceTurnId : undefined;
+  const composerBlocked = steering || (!steeringTurnId && (sending || conversationTurnRunning));
+  const quickPromptBlocked = steering || sending || conversationTurnRunning;
   const managerChatItems = useMemo(
     () => items.filter((item) => item.kind === "message"
       || (item.kind === "proposal" && (sessionProposalIds.includes(item.proposal.previewId)
@@ -1626,7 +1631,36 @@ export function PersonalWorkspacePage({
   async function sendMessage(messageOverride?: string) {
     const pendingImages = messageOverride ? [] : imageAttachments;
     const message = (messageOverride ?? composer).trim() || (pendingImages.length ? t("composer.imageAnalysisPrompt") : "");
-    if (!message || sending || conversationHistoryState?.sendBlocked) return;
+    if (!message || composerBlocked || conversationHistoryState?.sendBlocked) return;
+    const previousSteering = steeringRequests.current.get(composerDraftKey);
+    const retry = previousSteering && previousSteering.sessionId === conversationSessionId && previousSteering.text === message
+      ? previousSteering : undefined;
+    if ((retry || steeringTurnId) && conversationSessionId && callbacks.onSteerConversationTurn) {
+      if (pendingImages.length) {
+        setImageAttachmentError(locale === "zh-CN" ? "本轮追加指令暂不支持图片，图片和草稿已保留。" : "This turn accepts text instructions only. Images and draft retained.");
+        return;
+      }
+      const request = retry ?? { sessionId: conversationSessionId, turnId: steeringTurnId!, text: message, id: crypto.randomUUID() };
+      steeringRequests.current.set(composerDraftKey, request);
+      setSteering(true);
+      setActionFeedback(null);
+      setImageAttachmentError(null);
+      try {
+        await callbacks.onSteerConversationTurn(selectedGoalId ?? "manager", request.turnId, message, request.id);
+        steeringRequests.current.delete(composerDraftKey);
+        if (!messageOverride) setComposerDraft(composerDraftKey, "", composer);
+        setActionFeedback(locale === "zh-CN" ? "执行器已接收本轮追加指令。" : "The executor accepted instructions for this turn.");
+      } catch (error) {
+        // Unknown delivery retries the original Turn even after it completes.
+        // A confirmed non-delivery may use a new ingress after recovery.
+        if (error instanceof ChatApiError && error.payload.delivery_state === "not_delivered") {
+          steeringRequests.current.delete(composerDraftKey);
+        }
+        setActionFeedback(error instanceof Error ? error.message : t("feedback.sendGenericError"));
+      } finally { setSteering(false); }
+      return;
+    }
+    if (sending) return;
     followConversationRef.current = true;
     setShowLatestMessage(false);
     if (loopxMode?.session_id === conversationSessionId && loopxMode?.enabled && loopxMode.active_turn_id && conversationSessionId) {
@@ -1969,16 +2003,16 @@ export function PersonalWorkspacePage({
               <summary>{locale === "zh-CN" ? "快捷提问" : "Suggestions"}</summary>
             {selectedGoal ? (
               <div className="personal-quick-prompts">
-                <button aria-label={t("composer.nextAction")} disabled={composerBlocked} onClick={() => void sendMessage(t("composer.nextActionPrompt"))} title={t("composer.sendMessageHint")} type="button"><MessageCircleQuestion size={13} /><span>{t("composer.nextAction")}</span></button>
-                <button aria-label={t("composer.agentProgress")} disabled={composerBlocked} onClick={() => void sendMessage(t("composer.agentProgressPrompt"))} title={t("composer.sendMessageHint")} type="button"><Send size={13} /><span>{t("composer.agentProgress")}</span></button>
+                <button aria-label={t("composer.nextAction")} disabled={quickPromptBlocked} onClick={() => void sendMessage(t("composer.nextActionPrompt"))} title={t("composer.sendMessageHint")} type="button"><MessageCircleQuestion size={13} /><span>{t("composer.nextAction")}</span></button>
+                <button aria-label={t("composer.agentProgress")} disabled={quickPromptBlocked} onClick={() => void sendMessage(t("composer.agentProgressPrompt"))} title={t("composer.sendMessageHint")} type="button"><Send size={13} /><span>{t("composer.agentProgress")}</span></button>
                 <button aria-label={t("composer.monitor")} disabled={sending} onClick={() => prepareScheduleDraft("monitor", selectedGoalId)} title={t("composer.sendMessageHint")} type="button"><CalendarClock size={13} /><span>{t("composer.monitor")}</span></button>
-                <button aria-label={t("composer.blockers")} disabled={composerBlocked || !stewardPromptText("gate")} onClick={() => void sendMessage(stewardPromptText("gate"))} title={t("composer.sendMessageHint")} type="button"><AlertCircle size={13} /><span>{t("composer.blockers")}</span></button>
-                <button aria-label={t("composer.evidence")} disabled={composerBlocked || !stewardPromptText("evidence")} onClick={() => void sendMessage(stewardPromptText("evidence"))} title={t("composer.sendMessageHint")} type="button"><FileText size={13} /><span>{t("composer.evidence")}</span></button>
+                <button aria-label={t("composer.blockers")} disabled={quickPromptBlocked || !stewardPromptText("gate")} onClick={() => void sendMessage(stewardPromptText("gate"))} title={t("composer.sendMessageHint")} type="button"><AlertCircle size={13} /><span>{t("composer.blockers")}</span></button>
+                <button aria-label={t("composer.evidence")} disabled={quickPromptBlocked || !stewardPromptText("evidence")} onClick={() => void sendMessage(stewardPromptText("evidence"))} title={t("composer.sendMessageHint")} type="button"><FileText size={13} /><span>{t("composer.evidence")}</span></button>
               </div>
             ) : (
               <div className="personal-quick-prompts">
-                <button aria-label={t("composer.globalTasks")} disabled={composerBlocked} onClick={() => void sendMessage(t("composer.globalTasksPrompt"))} title={t("composer.sendMessageHint")} type="button"><MessageCircleQuestion size={13} /><span>{t("composer.globalTasks")}</span></button>
-                <button aria-label={t("composer.globalProgress")} disabled={composerBlocked} onClick={() => void sendMessage(t("composer.globalProgressPrompt"))} title={t("composer.sendMessageHint")} type="button"><Send size={13} /><span>{t("composer.globalProgress")}</span></button>
+                <button aria-label={t("composer.globalTasks")} disabled={quickPromptBlocked} onClick={() => void sendMessage(t("composer.globalTasksPrompt"))} title={t("composer.sendMessageHint")} type="button"><MessageCircleQuestion size={13} /><span>{t("composer.globalTasks")}</span></button>
+                <button aria-label={t("composer.globalProgress")} disabled={quickPromptBlocked} onClick={() => void sendMessage(t("composer.globalProgressPrompt"))} title={t("composer.sendMessageHint")} type="button"><Send size={13} /><span>{t("composer.globalProgress")}</span></button>
                 <button aria-label={t("composer.createGoal")} onClick={requestGoalCreate} title={t("composer.createGoalHint")} type="button"><Plus size={13} /><span>{t("composer.createGoal")}</span></button>
               </div>
             )}
@@ -1991,7 +2025,9 @@ export function PersonalWorkspacePage({
               </figure>
             ))}</div> : null}
             {imageAttachmentError ? <p className="personal-composer-error" role="alert">{imageAttachmentError}</p> : null}
-            {conversationTurnRunning ? <p className="personal-composer-status" role="status">{t("composer.turnRunning")}</p> : null}
+            {conversationTurnRunning ? <p className="personal-composer-status" role="status">{steeringTurnId
+              ? (locale === "zh-CN" ? "本轮进行中 · 发消息可调整当前工作" : "Turn in progress · send instructions to adjust this work")
+              : t("composer.turnRunning")}</p> : null}
             <div
               className="personal-channel-composer"
               onDragOver={(event) => {
@@ -2034,7 +2070,9 @@ export function PersonalWorkspacePage({
               />
               <button aria-label={t("composer.send")} disabled={(!composer.trim() && imageAttachments.length === 0) || composerBlocked || conversationHistoryState?.sendBlocked} onClick={() => void sendMessage()} title={t("composer.sendMessageHint")} type="button"><Send size={18} /></button>
             </div>
-            {conversationOpen ? <div className="personal-composer-hint">{sending
+            {conversationOpen ? <div className="personal-composer-hint">{steering
+              ? (locale === "zh-CN" ? "正在发送本轮追加指令…" : "Sending instructions for this turn…")
+              : sending && !steeringTurnId
               ? (locale === "zh-CN" ? "正在回复 · 修改当前任务请使用“调整本轮”" : "Reply in progress · use Adjust turn to change the current task")
               : (locale === "zh-CN" ? "Enter 发送 · Shift+Enter 换行" : "Enter to send · Shift+Enter for a new line")}</div> : null}
             </>}

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

@@ -56,6 +56,7 @@ import {
   resumeChatTurnStreaming,
   sendChatTurnStreaming,
   chatSessionQueuesFollowUps,
+  chatSessionSupportsSteering,
   selectAvailableChatAgent,
   sessionInvalidatedByPayload,
   todoNoWriteReceiptFromPayload,
@@ -1410,6 +1411,7 @@ function PersonalGoalHome({
   // Bound Sessions whose mode queues a message sent while a Turn runs, read
   // from the Session owner each time this page binds a Session.
   const [followUpQueueSessionIds, setFollowUpQueueSessionIds] = useState<ReadonlySet<string>>(() => new Set());
+  const [steeringSessionIds, setSteeringSessionIds] = useState<ReadonlySet<string>>(() => new Set());
   const [executionSessions, setExecutionSessions] = useState<ChatSessionSummary[]>([]);
   // Bumped when the service reports a running Turn this page did not know
   // about, so the Turn recovery effect re-reads the Session and adopts it.
@@ -1559,6 +1561,14 @@ function PersonalGoalHome({
 
   function recordSessionAdmission(session: ChatSessionSummary) {
     const queues = chatSessionQueuesFollowUps(session);
+    const supportsSteering = chatSessionSupportsSteering(session);
+    setSteeringSessionIds((current) => {
+      if (current.has(session.session_id) === supportsSteering) return current;
+      const next = new Set(current);
+      if (supportsSteering) next.add(session.session_id);
+      else next.delete(session.session_id);
+      return next;
+    });
     setFollowUpQueueSessionIds((current) => {
       if (current.has(session.session_id) === queues) return current;
       const next = new Set(current);
@@ -2939,9 +2949,9 @@ function PersonalGoalHome({
           },
           onSteerConversationTurn: async (targetContextId, turnId, message, ingressId) => {
             const binding = runtimeBindings[targetContextId];
-            if (!binding?.sessionId || binding.turnId !== turnId || activeTurnIds.current.get(targetContextId) !== turnId) {
-              throw new Error("本轮已结束或已被新的回合取代，追加指令未发送，草稿已保留。");
-            }
+            if (!binding?.sessionId) throw new Error("当前会话不可用，追加指令未发送，草稿已保留。");
+            // The service owns exact-turn admission and durable retry. A delivered
+            // ingress may be read back after completion; never retarget it locally.
             await steerChatTurn(binding.sessionId, turnId, message, ingressId);
             const id = managerMessageId.current++;
             setMessagesByContext(current => {
@@ -3081,6 +3091,7 @@ function PersonalGoalHome({
         managerRuntime={managerRuntime}
         conversationSessionId={runtimeBindings[contextId]?.sessionId}
         conversationQueuesFollowUps={followUpQueueSessionIds.has(runtimeBindings[contextId]?.sessionId ?? "")}
+        conversationSupportsSteering={steeringSessionIds.has(runtimeBindings[contextId]?.sessionId ?? "")}
         conversationHistoryState={conversationHistory}
         model={workspaceModel}
         readOnly={readOnly}

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -126,6 +126,16 @@ receipt → observed work or actionable failure → readable answer in the same
 - Before dispatch, cancel only session preparation and state that the request
   was not submitted. After acceptance, existing exact-turn steering/interrupt
   controls own effects; stopping observation is not stopping the worker.
+- During a managed Codex Turn, the ordinary composer sends text instructions to
+  that exact Turn, without a second adjustment form or another Turn submission.
+  Attached-host messages keep their next-Turn queue semantics; unsupported
+  managed adapters keep the draft without advertising native steering. The
+  idle composer and initial presentation are unchanged. This changes the former
+  managed-running Send lockout; LoopX mode retains its explicit delivery choice.
+  A lost or mismatched receipt preserves draft and ingress identity, including
+  retry after completion. Only a confirmed non-delivery permits a new ingress.
+  Acceptance means the executor received the instructions, not that it adopted
+  them or that delegated/team work stopped. Live adoption stays a release gate.
 - The compact receipt and full conversation offer the same controls. Failure
   ends the live indicator, preserves the request/partial answer and names the
   next supported action. A completed delegation still shows receiver adoption