fix: reliably submit delayed agent prompts (#3506)

2026-09-21 16:01:04 +00:00 · 2026-09-02 13:34:03 +02:00
parent 6c0bb273d5
commit 8633a398e6
16 changed files with 985 additions and 381 deletions
@@ -73,7 +73,7 @@ herdr agent rename w1:p2 reviewer

 Pane input addresses the terminal regardless of its current occupant. Agent input resolves the live agent and rejects the operation if that agent no longer controls the pane.

-`agent prompt --wait` submits immediately unless the agent is already `blocked`, in which case it returns `agent_blocked` without starting the wait. If an accepted prompt starts from another non-working state, Herdr must observe a lifecycle change within five seconds. Otherwise, it returns `agent_prompt_stalled` instead of waiting indefinitely; a caller `--timeout` of five seconds or less returns the normal `timeout` error. After Herdr observes activity, it waits for the requested settled status. It does not track individual turns. If the agent is already working, completion of that active turn may satisfy the wait. Standalone `agent wait` observes the current agent and returns immediately if its status already matches. Both commands default to `idle`, `done`, or `blocked`. Repeat `--until` to accept several exact states, for example `--until idle --until done`; use `--until unknown` explicitly when needed. On `agent prompt`, `--until` requires `--wait`.
+`agent prompt --wait` submits immediately unless the agent is already `blocked`, in which case it returns `agent_blocked` without starting the wait. If an accepted prompt starts from another non-working state, Herdr must observe `working` or `blocked` within five seconds. Otherwise, it returns `agent_prompt_stalled`; if the caller timeout expires first, it returns the normal `timeout` error. This prevents unrelated `idle`, `done`, or session changes from completing the wait. After Herdr observes activity, it waits for the requested settled status. It does not track individual turns. If the agent is already working, completion of that active turn may satisfy the wait. Standalone `agent wait` observes the current agent and returns immediately if its status already matches. Both commands default to `idle`, `done`, or `blocked`. Repeat `--until` to accept several exact states, for example `--until idle --until done`; use `--until unknown` explicitly when needed. On `agent prompt`, `--until` requires `--wait`.

 `idle` means the agent is ready for input and its tab has been seen in the focused Herdr UI. `done` is the same underlying idle state after background work finishes, until that tab is focused or `pane focus` / `agent focus` targets it. Reading through the CLI does not mark it seen. `blocked` means Herdr recognized an approval or question UI. `unknown` means an agent is present but Herdr cannot classify its lifecycle confidently; it does not prove successful completion. Use exact `--until` states when that distinction matters.

@@ -310,7 +310,7 @@ Agent targets are either a unique live agent name or the pane ID that currently

 A successful start returns only after the expected agent owns the same terminal and is ready for interactive input. If detection reports `blocked` during startup, the command returns `agent_not_ready` immediately. The name remains available for `agent read` and `agent send-keys`, and becomes ready for prompts after detection reports `idle`. The default startup timeout is 30000 milliseconds; explicit values must be greater than 3000 and no more than 300000.

-`agent prompt` honors live bracketed-paste mode and sends text followed by encoded Enter after a short delay, including while the agent is working. If the agent is already `blocked`, it returns `agent_blocked` without sending input. With `--wait`, an accepted prompt sent from another non-working state must produce an observed lifecycle change within five seconds or Herdr returns `agent_prompt_stalled`; a caller timeout of five seconds or less keeps the normal `timeout` error. After activity is observed, it waits for the first requested settled status. It does not track individual turns. If the agent is already working, completion of that active turn may satisfy the wait. `--until` narrows the matching states and is rejected unless `--wait` is also present. Standalone `agent wait` returns immediately when the current status matches. Both default to `idle`, `done`, or `blocked`; use `--until unknown` explicitly when needed.
+`agent prompt` honors live bracketed-paste mode and sends text followed by encoded Enter after a short delay, including while the agent is working. If the agent is already `blocked`, it returns `agent_blocked` without sending input. With `--wait`, an accepted prompt sent from another non-working state must produce an observed `working` or `blocked` state within five seconds or Herdr returns `agent_prompt_stalled`; if the caller timeout expires first, Herdr returns the normal `timeout` error. This prevents unrelated `idle`, `done`, or session changes from completing the wait. After activity is observed, it waits for the first requested settled status. It does not track individual turns. If the agent is already working, completion of that active turn may satisfy the wait. `--until` narrows the matching states and is rejected unless `--wait` is also present. Standalone `agent wait` returns immediately when the current status matches. Both default to `idle`, `done`, or `blocked`; use `--until unknown` explicitly when needed.

 `idle` means ready for input after its tab has been seen in the focused Herdr UI; `done` is the same underlying idle state after unseen background work completes. Focusing that tab or targeting it with `pane focus` / `agent focus` marks it seen; reading it through the CLI does not. `blocked` means Herdr recognized an approval or question UI. `unknown` means an agent is present but Herdr cannot classify it confidently, not that its work succeeded.

@@ -356,7 +356,7 @@ Use `pane wait-output` for normal commands and servers. Use `agent wait` for cod

 `pane wait-output` checks the selected snapshot immediately, including output that already exists, then polls until it matches. The default source name is `recent`; matching treats it as unwrapped recent output from the latest 80 rendered terminal rows. `--lines` changes that row limit. `--match` finds a literal substring on one line, and `--regex` uses Rust regular-expression syntax and also matches one line at a time.

-`pane wait-output`, `agent wait`, and `agent prompt --wait` wait indefinitely when `--timeout` is omitted. A timeout or server error is emitted as JSON on stderr with exit status 1. CLI usage errors exit with status 2.
+`pane wait-output` and `agent wait` wait indefinitely when `--timeout` is omitted. For `agent prompt --wait`, the settled-state wait is indefinite after activity is observed or when the prompt starts in `working`; a non-working prompt still returns `agent_prompt_stalled` after five seconds without observed activity. A timeout or server error is emitted as JSON on stderr with exit status 1. CLI usage errors exit with status 2.

 ## Integrations

@@ -73,7 +73,7 @@ herdr agent rename w1:p2 reviewer

 ペイン入力は、現在のプロセスに関係なくターミナルを指定します。エージェント入力はライブエージェントを解決し、そのエージェントがペインを制御しなくなっていれば操作を拒否します。

-`agent prompt --wait` は、エージェントがすでに `blocked` でない限り、プロンプトをすぐ送信します。すでに `blocked` の場合は待機を開始せずに `agent_blocked` を返します。受け付けたプロンプトが別の non-working 状態から始まる場合、Herdr はまず 5 秒以内にライフサイクル変化が観測されることを要求します。状態シーケンスが進まなければ、無期限に待たず `agent_prompt_stalled` を返します。呼び出し側の `--timeout` が 5 秒以下なら、通常の `timeout` エラーを返します。活動を観測した後、要求された安定状態を待ちます。個々のターンは追跡しません。エージェントがすでに working の場合、進行中ターンの完了が待機を満たすことがあります。単独の `agent wait` は現在のエージェントを監視し、すでに状態が一致していれば即座に返ります。どちらもデフォルトでは `idle`、`done`、`blocked` を待ちます。複数の正確な状態を許可するには、`--until idle --until done` のように `--until` を繰り返します。`unknown` が必要な場合は `--until unknown` を明示してください。`agent prompt` では `--until` に `--wait` が必要です。
+`agent prompt --wait` は、エージェントがすでに `blocked` でない限り、プロンプトをすぐ送信します。すでに `blocked` の場合は待機を開始せずに `agent_blocked` を返します。受け付けたプロンプトが別の non-working 状態から始まる場合、Herdr は 5 秒以内に `working` または `blocked` を観測する必要があります。観測できなければ `agent_prompt_stalled` を返し、呼び出し側のタイムアウトが先に切れた場合は通常の `timeout` エラーを返します。これにより、無関係な `idle`、`done`、またはセッションの変化によって待機が誤って完了することを防ぎます。活動を観測した後、要求された安定状態を待ちます。個々のターンは追跡しません。エージェントがすでに working の場合、進行中ターンの完了が待機を満たすことがあります。単独の `agent wait` は現在のエージェントを監視し、すでに状態が一致していれば即座に返ります。どちらもデフォルトでは `idle`、`done`、`blocked` を待ちます。複数の正確な状態を許可するには、`--until idle --until done` のように `--until` を繰り返します。`unknown` が必要な場合は `--until unknown` を明示してください。`agent prompt` では `--until` に `--wait` が必要です。

 `idle` は、エージェントが入力待ちで、そのタブがフォーカス中の Herdr UI ですでに表示済みであることを示します。`done` は同じ基礎的な idle 状態ですが、バックグラウンド作業が完了してから、そのタブをフォーカスするか `pane focus` / `agent focus` で対象にするまで使われます。CLI で読み取るだけでは表示済みになりません。`blocked` は承認または質問 UI を Herdr が認識した状態です。`unknown` はエージェントが存在するもののライフサイクルを確実に分類できない状態で、成功完了を意味しません。違いが重要なら正確な `--until` を指定してください。

@@ -290,7 +290,7 @@ herdr agent explain --file PATH --agent LABEL [--json|--verbose]

 成功した start は、期待したエージェントが同じターミナルを所有し、対話入力の準備ができてから返ります。起動中の検出状態が `blocked` の場合、コマンドは直ちに `agent_not_ready` を返します。名前は `agent read` と `agent send-keys` で引き続き使用でき、検出状態が `idle` になるとプロンプトを送信できるようになります。デフォルトの起動タイムアウトは 30000 ミリ秒で、明示する値は 3000 より大きく 300000 以下でなければなりません。

-`agent prompt` は現在の bracketed paste モードを尊重し、working 中でもテキストを送信してから短い遅延の後にエンコード済み Enter を送信します。エージェントがすでに `blocked` の場合は、入力を送信せずに `agent_blocked` を返します。`--wait` を使う場合、別の non-working 状態から受け付けたプロンプトは 5 秒以内に観測可能なライフサイクル変化を起こす必要があり、変化がなければ Herdr は `agent_prompt_stalled` を返します。呼び出し側のタイムアウトが 5 秒以下なら、通常の `timeout` エラーを維持します。活動を観測した後、要求された安定状態を待ちます。個々のターンは追跡しません。すでに working の場合、進行中ターンの完了が待機を満たすことがあります。`--until` は一致状態を絞り込み、`--wait` なしでは拒否されます。単独の `agent wait` は現在の状態が一致すれば即座に返ります。どちらもデフォルトは `idle`、`done`、`blocked` です。
+`agent prompt` は現在の bracketed paste モードを尊重し、working 中でもテキストを送信してから短い遅延の後にエンコード済み Enter を送信します。エージェントがすでに `blocked` の場合は、入力を送信せずに `agent_blocked` を返します。`--wait` を使う場合、別の non-working 状態から受け付けたプロンプトでは、5 秒以内に `working` または `blocked` が観測される必要があります。観測できなければ Herdr は `agent_prompt_stalled` を返し、呼び出し側のタイムアウトが先に切れた場合は通常の `timeout` エラーを返します。これにより、無関係な `idle`、`done`、またはセッションの変化によって待機が誤って完了することを防ぎます。活動を観測した後、要求された最初の安定状態を待ちます。個々のターンは追跡しません。すでに working の場合、進行中ターンの完了が待機を満たすことがあります。`--until` は一致状態を絞り込み、`--wait` なしでは拒否されます。単独の `agent wait` は現在の状態が一致すれば即座に返ります。どちらもデフォルトは `idle`、`done`、`blocked` です。

 `idle` は入力待ちで、そのタブがフォーカス中の Herdr UI ですでに表示済みの状態です。`done` は同じ基礎的な idle 状態ですが、表示していないバックグラウンド作業が完了したときに使われます。そのタブをフォーカスするか `pane focus` / `agent focus` で対象にすると表示済みになりますが、CLI で読み取るだけでは変わりません。`blocked` は承認または質問 UI を Herdr が認識した状態です。`unknown` はエージェントが存在するものの確実に分類できない状態で、作業の成功を意味しません。

@@ -323,7 +323,7 @@ herdr pane wait-output <pane_id> (--match <text> | --regex <pattern>) [--source

 `pane wait-output` は選択したスナップショットを最初にすぐ検索するため、すでに存在する出力も一致します。デフォルトのソース名は `recent` で、直近 80 行の描画済みターミナル行を折り返し前の出力として扱います。`--lines` でその行数を変更できます。`--match` は 1 行内のリテラル部分文字列を検索し、`--regex` は Rust の正規表現構文で 1 行ずつ一致します。

-`pane wait-output`、`agent wait`、`agent prompt --wait` は、`--timeout` を省略すると無期限に待ちます。タイムアウトまたはサーバーエラーは JSON として stderr に出力され、終了コード 1 になります。CLI 使用法のエラーは終了コード 2 です。
+`pane wait-output` と `agent wait` は、`--timeout` を省略すると無期限に待ちます。`agent prompt --wait` では、活動を観測した後、またはプロンプト開始時に `working` だった場合の安定状態待機は無期限です。non-working 状態から開始して活動を観測できない場合は、5 秒後に `agent_prompt_stalled` を返します。タイムアウトまたはサーバーエラーは JSON として stderr に出力され、終了コード 1 になります。CLI 使用法のエラーは終了コード 2 です。

 ## インテグレーション

@@ -73,7 +73,7 @@ herdr agent rename w1:p2 reviewer

 窗格输入直接指定终端,不关心当前进程。智能体输入会解析实时智能体;如果该智能体已不再控制此窗格,操作会被拒绝。

-除非智能体已经是 `blocked`,`agent prompt --wait` 会立即提交提示。已经是 `blocked` 时,它不会开始等待,而是返回 `agent_blocked`。已接受的提示从其他非 working 状态开始时,Herdr 首先要求在五秒内观察到生命周期变化。如果状态序列没有前进，它会返回 `agent_prompt_stalled`，而不是无限等待；调用方设置的 `--timeout` 不超过五秒时，仍返回普通的 `timeout` 错误。观察到活动后，它会等待请求的稳定状态。它不会跟踪单独的轮次。如果智能体已经处于 working，当前轮次的完成可能满足等待。独立的 `agent wait` 会观察当前智能体；如果状态已经匹配，就会立即返回。两者默认匹配 `idle`、`done` 或 `blocked`。可以重复使用 `--until` 接受多个精确状态，例如 `--until idle --until done`；需要 `unknown` 时请明确使用 `--until unknown`。在 `agent prompt` 中，`--until` 必须与 `--wait` 一起使用。
+除非智能体已经是 `blocked`,`agent prompt --wait` 会立即提交提示。已经是 `blocked` 时,它不会开始等待,而是返回 `agent_blocked`。已接受的提示从其他非 working 状态开始时,Herdr 必须在五秒内观察到 `working` 或 `blocked`。如果没有观察到，Herdr 返回 `agent_prompt_stalled`；如果调用方的超时先到期，则返回普通的 `timeout` 错误。这可防止无关的 `idle`、`done` 或会话变化错误地完成等待。观察到活动后，它会等待请求的稳定状态。它不会跟踪单独的轮次。如果智能体已经处于 working，当前轮次的完成可能满足等待。独立的 `agent wait` 会观察当前智能体；如果状态已经匹配，就会立即返回。两者默认匹配 `idle`、`done` 或 `blocked`。可以重复使用 `--until` 接受多个精确状态，例如 `--until idle --until done`；需要 `unknown` 时请明确使用 `--until unknown`。在 `agent prompt` 中，`--until` 必须与 `--wait` 一起使用。

 `idle` 表示智能体正等待输入,且其标签页已在聚焦的 Herdr 界面中显示。`done` 是相同的底层 idle 状态,但用于后台工作完成后,直到该标签页获得焦点或 `pane focus` / `agent focus` 指向它。仅通过 CLI 读取不会把它标记为已查看。`blocked` 表示 Herdr 识别到了审批或提问界面。`unknown` 表示智能体存在,但 Herdr 无法可靠判断其生命周期;它不代表工作成功完成。区别重要时,请指定精确的 `--until` 状态。

@@ -290,7 +290,7 @@ herdr agent explain --file PATH --agent LABEL [--json|--verbose]

 成功的 start 只有在预期智能体占用同一终端并可接受交互输入后才返回。如果启动期间检测到 `blocked`,命令会立即返回 `agent_not_ready`。该名称仍可用于 `agent read` 和 `agent send-keys`,检测变为 `idle` 后即可用于发送提示。默认启动超时是 30000 毫秒;显式值必须大于 3000 且不超过 300000。

-`agent prompt` 遵循当前的 bracketed paste 模式,即使智能体处于 working 也会先发送文本,短暂延迟后再发送编码后的 Enter。如果智能体已经是 `blocked`,它不会发送输入,而是返回 `agent_blocked`。使用 `--wait` 时,从其他非 working 状态接受的提示必须在五秒内产生可观察的生命周期变化，否则 Herdr 返回 `agent_prompt_stalled`；调用方超时不超过五秒时，仍返回普通的 `timeout` 错误。观察到活动后，它会等待请求的稳定状态。它不会跟踪单独的轮次。如果智能体已经处于 working，当前轮次的完成可能满足等待。`--until` 用于缩小匹配状态，不带 `--wait` 时会被拒绝。独立的 `agent wait` 在当前状态匹配时立即返回。两者默认匹配 `idle`、`done` 或 `blocked`；需要 `unknown` 时请明确使用 `--until unknown`。
+`agent prompt` 遵循当前的 bracketed paste 模式,即使智能体处于 working 也会先发送文本,短暂延迟后再发送编码后的 Enter。如果智能体已经是 `blocked`,它不会发送输入,而是返回 `agent_blocked`。使用 `--wait` 时,从其他非 working 状态接受的提示必须在五秒内产生可观察的 `working` 或 `blocked` 状态，否则 Herdr 返回 `agent_prompt_stalled`；如果调用方的超时先到期，则返回普通的 `timeout` 错误。这可防止无关的 `idle`、`done` 或会话变化错误地完成等待。观察到活动后，它会等待请求的第一个稳定状态。它不会跟踪单独的轮次。如果智能体已经处于 working，当前轮次的完成可能满足等待。`--until` 用于缩小匹配状态，不带 `--wait` 时会被拒绝。独立的 `agent wait` 在当前状态匹配时立即返回。两者默认匹配 `idle`、`done` 或 `blocked`；需要 `unknown` 时请明确使用 `--until unknown`。

 `idle` 表示智能体在等待输入,且其标签页已在聚焦的 Herdr 界面中显示。`done` 是相同的底层 idle 状态,用于未查看的后台工作完成后。聚焦该标签页或用 `pane focus` / `agent focus` 指向它会标记为已查看;仅通过 CLI 读取不会改变它。`blocked` 表示 Herdr 识别到审批或提问界面。`unknown` 表示智能体存在但无法可靠分类,不代表工作成功。

@@ -323,7 +323,7 @@ herdr pane wait-output <pane_id> (--match <text> | --regex <pattern>) [--source

 `pane wait-output` 会立即检查所选快照,所以已经存在的输出也能匹配。默认来源名是 `recent`;匹配时会把最近 80 个已渲染终端行作为未折行输出处理。`--lines` 可以修改这个行数限制。`--match` 在单行中查找字面子串;`--regex` 使用 Rust 正则表达式语法并逐行匹配。

-`pane wait-output`、`agent wait` 和 `agent prompt --wait` 在省略 `--timeout` 时会无限等待。超时或服务器错误以 JSON 写到 stderr,退出状态为 1。CLI 用法错误的退出状态为 2。
+`pane wait-output` 和 `agent wait` 在省略 `--timeout` 时会无限等待。对于 `agent prompt --wait`，观察到活动后，或提示开始时已处于 `working`，等待稳定状态的过程不会超时；如果提示从 non-working 状态开始且未观察到活动，五秒后仍会返回 `agent_prompt_stalled`。超时或服务器错误以 JSON 写到 stderr,退出状态为 1。CLI 用法错误的退出状态为 2。

 ## 集成

@@ -193,7 +193,6 @@ pub(super) fn prompt_agent(
        )));
    };

-    let last_event_sequence = event_hub.current_sequence();
    let before_prompt = match agent_get(&request_id, &params.target, api_tx) {
        Ok(agent) => agent,
        Err(response) => {
@@ -202,7 +201,11 @@ pub(super) fn prompt_agent(
                .map_err(std::io::Error::other);
        }
    };
+    let prompt_started_working =
+        before_prompt.agent_status == crate::api::schema::AgentStatus::Working;
    let target = params.target.clone();
+    let wait_started = std::time::Instant::now();
+    let last_event_sequence = event_hub.current_sequence();
    let prompt_response = dispatch_to_app_with_timeout(
        Request {
            id: request_id.clone(),
@@ -223,39 +226,38 @@ pub(super) fn prompt_agent(
        return agent_wait_not_running(request_id).map(Some);
    }

-    let wait_started = std::time::Instant::now();
+    let prompt_activity_observed = prompt_started_working
+        || matches!(
+            prompted.agent_status,
+            crate::api::schema::AgentStatus::Working | crate::api::schema::AgentStatus::Blocked
+        );
    let prompt_state_change_seq = prompted.state_change_seq;
    let until = agent_wait_statuses(wait.until);
    let mut initial = prompted;
-    let mut after_state_change_seq = Some(prompt_state_change_seq);

-    if initial.agent_status != crate::api::schema::AgentStatus::Working {
-        let effect_timeout_ms = wait
-            .timeout_ms
-            .map_or(AGENT_PROMPT_EFFECT_TIMEOUT_MS, |timeout_ms| {
-                timeout_ms.min(AGENT_PROMPT_EFFECT_TIMEOUT_MS)
-            });
-        let timeout_kind = if wait
-            .timeout_ms
-            .is_some_and(|timeout_ms| timeout_ms <= AGENT_PROMPT_EFFECT_TIMEOUT_MS)
-        {
-            AgentWaitTimeoutKind::Status
-        } else {
-            AgentWaitTimeoutKind::PromptStalled {
-                baseline: prompt_state_change_seq,
-                timeout_ms: effect_timeout_ms,
+    if !prompt_activity_observed {
+        let remaining_timeout_ms = remaining_timeout_ms(wait.timeout_ms, wait_started);
+        let (effect_timeout_ms, timeout_kind) = match remaining_timeout_ms {
+            Some(timeout_ms) if timeout_ms <= AGENT_PROMPT_EFFECT_TIMEOUT_MS => {
+                (timeout_ms, AgentWaitTimeoutKind::Status)
            }
+            _ => (
+                AGENT_PROMPT_EFFECT_TIMEOUT_MS,
+                AgentWaitTimeoutKind::PromptStalled {
+                    timeout_ms: AGENT_PROMPT_EFFECT_TIMEOUT_MS,
+                },
+            ),
        };
        let Some(outcome) = wait_for_resolved_agent(
            request_id.clone(),
            ResolvedAgentWait {
                target: target.clone(),
-                until: all_agent_statuses(),
+                until: prompt_activity_statuses(),
                timeout_ms: Some(effect_timeout_ms),
                initial,
                last_event_sequence,
-                after_state_change_seq,
-                accept_transient_status: false,
+                after_state_change_seq: Some(prompt_state_change_seq),
+                accept_transient_status: true,
                timeout_kind,
            },
            stream,
@@ -270,10 +272,9 @@ pub(super) fn prompt_agent(
            AgentWaitOutcome::Matched(agent) => *agent,
            AgentWaitOutcome::Response(response) => return Ok(Some(response)),
        };
-        after_state_change_seq = None;
-        if agent_wait_matches(&initial, &until, None) {
-            return agent_prompt_success(request_id, initial).map(Some);
-        }
+    }
+    if agent_wait_matches(&initial, &until, None) {
+        return agent_prompt_success(request_id, initial).map(Some);
    }

    let Some(outcome) = wait_for_resolved_agent(
@@ -286,7 +287,7 @@ pub(super) fn prompt_agent(
            // Replay from before submission so terminal lifecycle events consumed by
            // the activity gate still terminate this settled-state wait.
            last_event_sequence,
-            after_state_change_seq,
+            after_state_change_seq: None,
            accept_transient_status: false,
            timeout_kind: AgentWaitTimeoutKind::Status,
        },
@@ -337,7 +338,7 @@ struct ResolvedAgentWait {
 #[derive(Clone, Copy)]
 enum AgentWaitTimeoutKind {
    Status,
-    PromptStalled { baseline: u64, timeout_ms: u64 },
+    PromptStalled { timeout_ms: u64 },
 }

 enum AgentWaitOutcome {
@@ -497,14 +498,10 @@ fn wait_for_resolved_agent(
    }
 }

-fn all_agent_statuses() -> Vec<crate::api::schema::AgentStatus> {
-    // Keep this exhaustive: every status is evidence that the sequence advanced.
+fn prompt_activity_statuses() -> Vec<crate::api::schema::AgentStatus> {
    vec![
-        crate::api::schema::AgentStatus::Idle,
        crate::api::schema::AgentStatus::Working,
        crate::api::schema::AgentStatus::Blocked,
-        crate::api::schema::AgentStatus::Done,
-        crate::api::schema::AgentStatus::Unknown,
    ]
 }

@@ -617,15 +614,12 @@ fn agent_wait_timeout(
        AgentWaitTimeoutKind::Status => {
            ("timeout", "timed out waiting for agent status".to_string())
        }
-        AgentWaitTimeoutKind::PromptStalled {
-            baseline,
-            timeout_ms,
-        } => {
+        AgentWaitTimeoutKind::PromptStalled { timeout_ms } => {
            let status = format!("{:?}", current.agent_status).to_ascii_lowercase();
            (
                "agent_prompt_stalled",
                format!(
-                    "agent prompt produced no observed state change within {timeout_ms} ms; status is {status} and state_change_seq remained {baseline}"
+                    "agent prompt produced no observed working or blocked state within {timeout_ms} ms; current status is {status}"
                ),
            )
        }
@@ -945,7 +945,13 @@ impl App {
                return self.handle_agent_view_clear(request.id, params)
            }
            Method::AgentStart(params) => return self.handle_agent_start(request.id, params),
-            Method::AgentPrompt(params) => return self.handle_agent_prompt(request.id, params),
+            Method::AgentPrompt(_) => {
+                return responses::encode_error(
+                    request.id,
+                    "invalid_request",
+                    "agent.prompt is handled asynchronously by the app runtime",
+                );
+            }
            Method::AgentWait(_) => {
                return responses::encode_error(
                    request.id,
@@ -59,13 +59,54 @@ impl App {
        encode_success(id, ResponseResult::AgentStarted { agent, argv })
    }

-    pub(super) fn handle_agent_prompt(&mut self, id: String, params: AgentPromptParams) -> String {
+    pub(crate) fn handle_deferred_agent_api_request(
+        &mut self,
+        request: crate::api::schema::Request,
+        respond_to: std::sync::mpsc::Sender<String>,
+    ) -> bool {
+        let crate::api::schema::Method::AgentPrompt(params) = request.method else {
+            return false;
+        };
+        match self.queue_agent_prompt(request.id, params) {
+            Ok((id, agent, completion)) => {
+                std::thread::spawn(move || {
+                    let response = match completion.recv() {
+                        Ok(Ok(())) => encode_success(id, ResponseResult::AgentPrompted { agent }),
+                        Ok(Err(err)) => encode_error(id, "agent_prompt_failed", err.to_string()),
+                        Err(_) => encode_error(id, "agent_prompt_failed", "pty actor closed"),
+                    };
+                    let _ = respond_to.send(response);
+                });
+            }
+            Err(response) => {
+                let _ = respond_to.send(response);
+            }
+        }
+        true
+    }
+
+    fn queue_agent_prompt(
+        &mut self,
+        id: String,
+        params: AgentPromptParams,
+    ) -> Result<
+        (
+            String,
+            crate::api::schema::AgentInfo,
+            std::sync::mpsc::Receiver<std::io::Result<()>>,
+        ),
+        String,
+    > {
        if params.text.is_empty() {
-            return encode_error(id, "empty_agent_prompt", "agent prompt must not be empty");
+            return Err(encode_error(
+                id,
+                "empty_agent_prompt",
+                "agent prompt must not be empty",
+            ));
        }
        let resolved = match self.resolve_agent_target(&params.target) {
            Ok(resolved) => resolved,
-            Err(err) => return encode_error_body(id, self.agent_target_error_body(err)),
+            Err(err) => return Err(encode_error_body(id, self.agent_target_error_body(err))),
        };
        let Some(terminal_id) = self
            .state
@@ -74,60 +115,65 @@ impl App {
            .and_then(|workspace| workspace.terminal_id(resolved.pane_id))
            .cloned()
        else {
-            return agent_not_found(id, &params.target);
+            return Err(agent_not_found(id, &params.target));
        };
        let Some(terminal) = self.state.terminals.get(&terminal_id) else {
-            return agent_not_found(id, &params.target);
+            return Err(agent_not_found(id, &params.target));
        };
        if terminal.state == crate::detect::AgentState::Blocked {
-            return encode_error(
+            return Err(encode_error(
                id,
                "agent_blocked",
                format!(
                    "agent {} is blocked and requires interactive input",
                    params.target
                ),
-            );
+            ));
        }
        let Some(expected_agent) = terminal.effective_known_agent() else {
-            return agent_not_ready(id, &params.target);
+            return Err(agent_not_ready(id, &params.target));
        };
        if terminal.managed_agent_launch_pending() {
-            return agent_not_ready(id, &params.target);
+            return Err(agent_not_ready(id, &params.target));
        }
        let Some(runtime) = self.lookup_runtime_sender(resolved.ws_idx, resolved.pane_id) else {
-            return agent_not_found(id, &params.target);
+            return Err(agent_not_found(id, &params.target));
        };
        if !super::super::agents::runtime_hosts_agent(runtime, expected_agent) {
-            return encode_error(
+            return Err(encode_error(
                id,
                "agent_not_ready",
                format!(
                    "agent {} is no longer the pane foreground process",
                    params.target
                ),
-            );
+            ));
        }
        if expected_agent == crate::detect::Agent::GithubCopilot {
            // Copilot ignores synthetic Enter after focus loss until it receives focus gained.
            let focus = match crate::ghostty::encode_focus(crate::ghostty::FocusEvent::Gained) {
                Ok(focus) => focus,
-                Err(err) => return encode_error(id, "agent_prompt_failed", err.to_string()),
+                Err(err) => {
+                    return Err(encode_error(id, "agent_prompt_failed", err.to_string()));
+                }
            };
            if let Err(err) = runtime.try_send_bytes(Bytes::from(focus)) {
-                return encode_error(id, "agent_prompt_failed", err.to_string());
+                return Err(encode_error(id, "agent_prompt_failed", err.to_string()));
            }
        }
        let (text, enter) =
            crate::app::api_helpers::encode_api_submission_parts(runtime, &params.text);
-        if let Err(err) = runtime.try_send_bytes(Bytes::from(text)) {
-            return encode_error(id, "agent_prompt_failed", err.to_string());
-        }
-        runtime.send_bytes_after(Bytes::from(enter), AGENT_PROMPT_SUBMIT_DELAY);
        let Some(agent) = self.agent_info(resolved.ws_idx, resolved.pane_id) else {
-            return agent_not_found(id, &params.target);
+            return Err(agent_not_found(id, &params.target));
        };
-        encode_success(id, ResponseResult::AgentPrompted { agent })
+        let completion = runtime
+            .queue_user_input_submission(
+                Bytes::from(text),
+                Bytes::from(enter),
+                AGENT_PROMPT_SUBMIT_DELAY,
+            )
+            .map_err(|err| encode_error(id.clone(), "agent_prompt_failed", err.to_string()))?;
+        Ok((id, agent, completion))
    }

    pub(super) fn handle_agent_read(
@@ -332,6 +378,28 @@ mod tests {
        app
    }

+    fn start_deferred_agent_prompt(
+        app: &mut App,
+        id: &str,
+        params: AgentPromptParams,
+    ) -> std::sync::mpsc::Receiver<String> {
+        let (respond_to, response_rx) = std::sync::mpsc::channel();
+        assert!(app.handle_deferred_agent_api_request(
+            crate::api::schema::Request {
+                id: id.into(),
+                method: crate::api::schema::Method::AgentPrompt(params),
+            },
+            respond_to,
+        ));
+        response_rx
+    }
+
+    fn run_deferred_agent_prompt(app: &mut App, id: &str, params: AgentPromptParams) -> String {
+        start_deferred_agent_prompt(app, id, params)
+            .recv_timeout(Duration::from_secs(1))
+            .expect("agent prompt responds after submission")
+    }
+
    #[tokio::test]
    async fn agent_prompt_sends_text_then_delays_enter() {
        let mut app = app_with_agent();
@@ -344,21 +412,26 @@ mod tests {
        terminal.set_detected_state(Some(Agent::OpenCode), AgentState::Working);
        let (runtime, mut rx) =
            crate::terminal::TerminalRuntime::test_with_channel_and_scrollback_bytes(
-                80, 24, 0, b"", 1,
+                80, 24, 0, b"", 2,
            );
        runtime.test_process_pty_bytes(b"\x1b[?2004h");
        app.state.insert_test_runtime(pane_id, runtime);

        let public_pane_id = app.public_pane_id(0, pane_id).unwrap();
        let bracketed_started = std::time::Instant::now();
-        let response = app.handle_agent_prompt(
-            "req".into(),
+        let response_rx = start_deferred_agent_prompt(
+            &mut app,
+            "req",
            AgentPromptParams {
                target: public_pane_id,
                text: "A != B".into(),
                wait: None,
            },
        );
+        assert!(response_rx.try_recv().is_err());
+        let response = response_rx
+            .recv_timeout(Duration::from_secs(1))
+            .expect("agent prompt responds after submission");
        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
        let ResponseResult::AgentPrompted { agent, .. } = success.result else {
            panic!("expected prompted response");
@@ -368,22 +441,16 @@ mod tests {
            rx.try_recv().unwrap(),
            Bytes::from_static(b"\x1b[200~A != B\x1b[201~")
        );
-        assert!(rx.try_recv().is_err());
-        assert_eq!(
-            tokio::time::timeout(Duration::from_secs(1), rx.recv())
-                .await
-                .unwrap()
-                .unwrap(),
-            Bytes::from_static(b"\r")
-        );
+        assert_eq!(rx.try_recv().unwrap(), Bytes::from_static(b"\r"));
        assert!(bracketed_started.elapsed() >= AGENT_PROMPT_SUBMIT_DELAY);

        app.lookup_runtime_sender(0, pane_id)
            .unwrap()
            .test_process_pty_bytes(b"\x1b[?2004l");
        let raw_started = std::time::Instant::now();
-        let raw = app.handle_agent_prompt(
-            "req-raw".into(),
+        let raw = run_deferred_agent_prompt(
+            &mut app,
+            "req-raw",
            AgentPromptParams {
                target: "reviewer".into(),
                text: "A != B".into(),
@@ -393,18 +460,12 @@ mod tests {
        let raw: SuccessResponse = serde_json::from_str(&raw).unwrap();
        assert!(matches!(raw.result, ResponseResult::AgentPrompted { .. }));
        assert_eq!(rx.try_recv().unwrap(), Bytes::from_static(b"A != B"));
-        assert!(rx.try_recv().is_err());
-        assert_eq!(
-            tokio::time::timeout(Duration::from_secs(1), rx.recv())
-                .await
-                .unwrap()
-                .unwrap(),
-            Bytes::from_static(b"\r")
-        );
+        assert_eq!(rx.try_recv().unwrap(), Bytes::from_static(b"\r"));
        assert!(raw_started.elapsed() >= AGENT_PROMPT_SUBMIT_DELAY);

-        let rejected = app.handle_agent_prompt(
-            "req-label".into(),
+        let rejected = run_deferred_agent_prompt(
+            &mut app,
+            "req-label",
            AgentPromptParams {
                target: "opencode".into(),
                text: "wrong target".into(),
@@ -429,8 +490,9 @@ mod tests {
        let (runtime, mut rx) = crate::terminal::TerminalRuntime::test_with_channel(80, 24);
        app.state.insert_test_runtime(pane_id, runtime);

-        let response = app.handle_agent_prompt(
-            "req".into(),
+        let response = run_deferred_agent_prompt(
+            &mut app,
+            "req",
            AgentPromptParams {
                target: "reviewer".into(),
                text: "unrelated prompt".into(),
@@ -468,8 +530,9 @@ mod tests {
        runtime.test_process_pty_bytes(b"\x1b[?2004h");
        app.state.insert_test_runtime(pane_id, runtime);

-        let response = app.handle_agent_prompt(
-            "req".into(),
+        let response = run_deferred_agent_prompt(
+            &mut app,
+            "req",
            AgentPromptParams {
                target: "reviewer".into(),
                text: "A != B".into(),
@@ -486,13 +549,7 @@ mod tests {
            rx.try_recv().unwrap(),
            Bytes::from_static(b"\x1b[200~A != B\x1b[201~")
        );
-        assert_eq!(
-            tokio::time::timeout(Duration::from_secs(1), rx.recv())
-                .await
-                .unwrap()
-                .unwrap(),
-            Bytes::from_static(b"\r")
-        );
+        assert_eq!(rx.try_recv().unwrap(), Bytes::from_static(b"\r"));
    }

    #[tokio::test]
@@ -552,8 +609,9 @@ mod tests {
        let (runtime, mut rx) = crate::terminal::TerminalRuntime::test_with_channel(80, 24);
        app.state.insert_test_runtime(pane_id, runtime);

-        let response = app.handle_agent_prompt(
-            "req-pending".into(),
+        let response = run_deferred_agent_prompt(
+            &mut app,
+            "req-pending",
            AgentPromptParams {
                target: "reviewer".into(),
                text: "A != B".into(),
@@ -367,7 +367,7 @@ fn agent_command() -> Command {
                        .help("Fail after this many milliseconds"),
                )
                .after_help(
-                    "If the agent is already blocked, submission is rejected with agent_blocked before any input is sent. When an accepted submission starts from another non-working state, --wait first requires an observed state change within 5000ms; otherwise it returns agent_prompt_stalled. A shorter --timeout returns timeout instead. It then matches idle, done, or blocked by default, or any exact --until state. It does not track turns: if the agent is already working, that active turn's completion may match. Without --timeout, the settled-state wait is indefinite.",
+                    "If the agent is already blocked, submission is rejected with agent_blocked before any input is sent. When an accepted submission starts from another non-working state, --wait requires an observed working or blocked state within 5000ms; otherwise it returns agent_prompt_stalled. A caller timeout that expires first returns timeout. It then matches idle, done, or blocked by default, or any exact --until state. It does not track turns: if the agent is already working, that active turn's completion may match.",
                ),
        )
        .subcommand(
@@ -1181,24 +1181,29 @@ impl PaneRuntimeIo {
        }
    }

-    fn send_bytes_after(&self, bytes: Bytes, delay: std::time::Duration) {
+    fn queue_user_input_submission(
+        &self,
+        text: Bytes,
+        enter: Bytes,
+        delay: std::time::Duration,
+    ) -> std::io::Result<std::sync::mpsc::Receiver<std::io::Result<()>>> {
        match self {
-            PaneRuntimeIo::Actor(actor) => {
-                let actor = actor.clone();
-                tokio::spawn(async move {
-                    tokio::time::sleep(delay).await;
-                    if let Err(err) = actor.write_user_input(bytes).await {
-                        warn!(error = %err, "failed to send delayed PTY input");
-                    }
-                });
-            }
+            PaneRuntimeIo::Actor(actor) => actor.queue_user_input_submission(text, enter, delay),
            #[cfg(test)]
            PaneRuntimeIo::TestChannel { sender, .. } => {
                let sender = sender.clone();
-                tokio::spawn(async move {
-                    tokio::time::sleep(delay).await;
-                    let _ = sender.send(bytes).await;
+                let (reply_tx, reply_rx) = std::sync::mpsc::channel();
+                std::thread::spawn(move || {
+                    let result = sender
+                        .try_send(text)
+                        .map_err(std::io::Error::other)
+                        .and_then(|()| {
+                            std::thread::sleep(delay);
+                            sender.try_send(enter).map_err(std::io::Error::other)
+                        });
+                    let _ = reply_tx.send(result);
                });
+                Ok(reply_rx)
            }
        }
    }
@@ -2856,8 +2861,13 @@ impl PaneRuntime {
        self.io.try_send_bytes(bytes)
    }

-    pub fn send_bytes_after(&self, bytes: Bytes, delay: std::time::Duration) {
-        self.io.send_bytes_after(bytes, delay);
+    pub fn queue_user_input_submission(
+        &self,
+        text: Bytes,
+        enter: Bytes,
+        delay: std::time::Duration,
+    ) -> std::io::Result<std::sync::mpsc::Receiver<std::io::Result<()>>> {
+        self.io.queue_user_input_submission(text, enter, delay)
    }

    pub fn try_send_paste(&self, text: String) -> Result<(), mpsc::error::TrySendError<Bytes>> {
@@ -43,6 +43,24 @@ mod windows {
        pub on_reader_exit: Option<ReaderExitCallback>,
    }

+    enum PtyIoDataCommand {
+        WriteUserInput(Bytes),
+        SubmitUserInput {
+            text: Bytes,
+            enter: Bytes,
+            delay: Duration,
+            reply: std_mpsc::Sender<std::io::Result<()>>,
+        },
+    }
+
+    enum PtyIoWriteCommand {
+        Write(Bytes),
+        SubmissionPart {
+            bytes: Bytes,
+            reply: std_mpsc::Sender<std::io::Result<()>>,
+        },
+    }
+
    enum PtyIoControlCommand {
        Resize(PtyResizeRequest),
        Shutdown,
@@ -50,28 +68,14 @@ mod windows {

    #[derive(Clone)]
    pub(crate) struct PtyIoActorHandle {
-        data_tx: mpsc::Sender<Bytes>,
+        data_tx: mpsc::Sender<PtyIoDataCommand>,
        control_tx: std_mpsc::Sender<PtyIoControlCommand>,
-        write_tx: std_mpsc::Sender<Bytes>,
+        write_tx: std_mpsc::Sender<PtyIoWriteCommand>,
        response_order: Arc<Mutex<()>>,
        accepting: Arc<Mutex<bool>>,
    }

    impl PtyIoActorHandle {
-        pub(crate) async fn write_user_input(
-            &self,
-            bytes: Bytes,
-        ) -> Result<(), mpsc::error::SendError<Bytes>> {
-            if !*self
-                .accepting
-                .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner())
-            {
-                return Err(mpsc::error::SendError(bytes));
-            }
-            self.data_tx.send(bytes).await
-        }
-
        pub(crate) fn try_write_user_input(
            &self,
            bytes: Bytes,
@@ -83,7 +87,58 @@ mod windows {
            {
                return Err(mpsc::error::TrySendError::Closed(bytes));
            }
-            self.data_tx.try_send(bytes)
+            self.data_tx
+                .try_send(PtyIoDataCommand::WriteUserInput(bytes))
+                .map_err(|err| match err {
+                    mpsc::error::TrySendError::Full(command) => {
+                        let PtyIoDataCommand::WriteUserInput(bytes) = command else {
+                            unreachable!("queued write returned another command")
+                        };
+                        mpsc::error::TrySendError::Full(bytes)
+                    }
+                    mpsc::error::TrySendError::Closed(command) => {
+                        let PtyIoDataCommand::WriteUserInput(bytes) = command else {
+                            unreachable!("queued write returned another command")
+                        };
+                        mpsc::error::TrySendError::Closed(bytes)
+                    }
+                })
+        }
+
+        pub(crate) fn queue_user_input_submission(
+            &self,
+            text: Bytes,
+            enter: Bytes,
+            delay: Duration,
+        ) -> std::io::Result<std_mpsc::Receiver<std::io::Result<()>>> {
+            let accepting = self
+                .accepting
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            if !*accepting {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::BrokenPipe,
+                    "pty actor closed",
+                ));
+            }
+            let (reply_tx, reply_rx) = std_mpsc::channel();
+            self.data_tx
+                .try_send(PtyIoDataCommand::SubmitUserInput {
+                    text,
+                    enter,
+                    delay,
+                    reply: reply_tx,
+                })
+                .map_err(|err| match err {
+                    mpsc::error::TrySendError::Full(_) => std::io::Error::new(
+                        std::io::ErrorKind::WouldBlock,
+                        "pty input queue is full",
+                    ),
+                    mpsc::error::TrySendError::Closed(_) => {
+                        std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed")
+                    }
+                })?;
+            Ok(reply_rx)
        }

        pub(crate) fn write_terminal_response(&self, response: impl FnOnce() -> Option<Bytes>) {
@@ -92,7 +147,7 @@ mod windows {
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            if let Some(bytes) = response().filter(|bytes| !bytes.is_empty()) {
-                let _ = self.write_tx.send(bytes);
+                let _ = self.write_tx.send(PtyIoWriteCommand::Write(bytes));
            }
        }

@@ -143,29 +198,22 @@ mod windows {
            let mut writer = master
                .take_writer()
                .map_err(|err| std::io::Error::other(err.to_string()))?;
-            let (data_tx, mut data_rx) = mpsc::channel::<Bytes>(1024);
+            let (data_tx, mut data_rx) = mpsc::channel::<PtyIoDataCommand>(1024);
            let (control_tx, control_rx) = std_mpsc::channel::<PtyIoControlCommand>();
-            let (write_tx, write_rx) = std_mpsc::channel::<Bytes>();
+            let (write_tx, write_rx) = std_mpsc::channel::<PtyIoWriteCommand>();
            let response_order = Arc::new(Mutex::new(()));
            let accepting = Arc::new(Mutex::new(!initially_quiesced));

            std::thread::spawn(move || {
-                for bytes in write_rx {
-                    if writer.write_all(&bytes).is_err() || writer.flush().is_err() {
-                        break;
-                    }
-                }
+                run_writer(&mut writer, write_rx);
                debug!(pane_id, "windows pty writer thread exiting");
            });

            {
                let write_tx = write_tx.clone();
+                let accepting = Arc::clone(&accepting);
                std::thread::spawn(move || {
-                    while let Some(bytes) = data_rx.blocking_recv() {
-                        if write_tx.send(bytes).is_err() {
-                            break;
-                        }
-                    }
+                    run_input_forwarder(&mut data_rx, write_tx, accepting);
                    debug!(pane_id, "windows pty input thread exiting");
                });
            }
@@ -183,11 +231,9 @@ mod windows {
                                    .lock()
                                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                                let result = on_read(&buf[..n]);
-                                if result
-                                    .terminal_responses
-                                    .into_iter()
-                                    .any(|response| write_tx.send(response).is_err())
-                                {
+                                if result.terminal_responses.into_iter().any(|response| {
+                                    write_tx.send(PtyIoWriteCommand::Write(response)).is_err()
+                                }) {
                                    break;
                                }
                            }
@@ -219,11 +265,9 @@ mod windows {
                                }) {
                                    warn!(pane_id, err = %err, "windows pty resize failed");
                                }
-                                if request
-                                    .terminal_responses
-                                    .into_iter()
-                                    .any(|response| write_tx.send(response).is_err())
-                                {
+                                if request.terminal_responses.into_iter().any(|response| {
+                                    write_tx.send(PtyIoWriteCommand::Write(response)).is_err()
+                                }) {
                                    break;
                                }
                            }
@@ -244,8 +288,207 @@ mod windows {
        }
    }

-    #[allow(dead_code)]
-    fn _assert_duration_send(_: Duration) {}
+    fn run_writer(writer: &mut impl Write, write_rx: std_mpsc::Receiver<PtyIoWriteCommand>) {
+        for command in write_rx {
+            let result = match command {
+                PtyIoWriteCommand::Write(bytes) => write_and_flush(writer, &bytes),
+                PtyIoWriteCommand::SubmissionPart { bytes, reply } => {
+                    let result = write_and_flush(writer, &bytes);
+                    let failed = result.is_err();
+                    let _ = reply.send(result);
+                    if failed {
+                        break;
+                    }
+                    continue;
+                }
+            };
+            if result.is_err() {
+                break;
+            }
+        }
+    }
+
+    fn run_input_forwarder(
+        data_rx: &mut mpsc::Receiver<PtyIoDataCommand>,
+        write_tx: std_mpsc::Sender<PtyIoWriteCommand>,
+        accepting: Arc<Mutex<bool>>,
+    ) {
+        while let Some(command) = data_rx.blocking_recv() {
+            match command {
+                PtyIoDataCommand::WriteUserInput(bytes) => {
+                    if write_tx.send(PtyIoWriteCommand::Write(bytes)).is_err() {
+                        break;
+                    }
+                }
+                PtyIoDataCommand::SubmitUserInput {
+                    text,
+                    enter,
+                    delay,
+                    reply,
+                } => {
+                    let result = write_submission_part(&write_tx, text).and_then(|()| {
+                        std::thread::sleep(delay);
+                        let accepting = accepting
+                            .lock()
+                            .unwrap_or_else(|poisoned| poisoned.into_inner());
+                        if !*accepting {
+                            return Err(pty_actor_closed());
+                        }
+                        write_submission_part(&write_tx, enter)
+                    });
+                    let failed = result.is_err();
+                    let _ = reply.send(result);
+                    if failed {
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    fn write_submission_part(
+        write_tx: &std_mpsc::Sender<PtyIoWriteCommand>,
+        bytes: Bytes,
+    ) -> std::io::Result<()> {
+        let (reply, completion) = std_mpsc::channel();
+        write_tx
+            .send(PtyIoWriteCommand::SubmissionPart { bytes, reply })
+            .map_err(|_| pty_actor_closed())?;
+        completion
+            .recv()
+            .unwrap_or_else(|_| Err(pty_actor_closed()))
+    }
+
+    fn pty_actor_closed() -> std::io::Error {
+        std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed")
+    }
+
+    fn write_and_flush(writer: &mut impl Write, bytes: &[u8]) -> std::io::Result<()> {
+        writer.write_all(bytes)?;
+        writer.flush()
+    }
+
+    #[cfg(test)]
+    mod tests {
+        use super::*;
+        use std::time::Instant;
+
+        struct RecordingWriter {
+            writes: Vec<(Vec<u8>, Instant)>,
+            flushes: Vec<Instant>,
+            fail_after: Option<usize>,
+            flushed: std_mpsc::Sender<()>,
+        }
+
+        impl Write for RecordingWriter {
+            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+                if self.fail_after == Some(self.writes.len()) {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::BrokenPipe,
+                        "writer closed",
+                    ));
+                }
+                self.writes.push((bytes.to_vec(), Instant::now()));
+                Ok(bytes.len())
+            }
+
+            fn flush(&mut self) -> std::io::Result<()> {
+                self.flushes.push(Instant::now());
+                let _ = self.flushed.send(());
+                Ok(())
+            }
+        }
+
+        fn run_recorded_submission(
+            fail_after: Option<usize>,
+            delay: Duration,
+            during_delay: impl FnOnce(&std_mpsc::Sender<PtyIoWriteCommand>, &Arc<Mutex<bool>>),
+        ) -> (RecordingWriter, std::io::Result<()>) {
+            let (flushed_tx, flushed_rx) = std_mpsc::channel();
+            let mut writer = RecordingWriter {
+                writes: Vec::new(),
+                flushes: Vec::new(),
+                fail_after,
+                flushed: flushed_tx,
+            };
+            let (data_tx, mut data_rx) = mpsc::channel(2);
+            let (write_tx, write_rx) = std_mpsc::channel();
+            let (reply_tx, reply_rx) = std_mpsc::channel();
+            let accepting = Arc::new(Mutex::new(true));
+            data_tx
+                .try_send(PtyIoDataCommand::SubmitUserInput {
+                    text: Bytes::from_static(b"prompt"),
+                    enter: Bytes::from_static(b"\r"),
+                    delay,
+                    reply: reply_tx,
+                })
+                .unwrap();
+            data_tx
+                .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
+                    b"user",
+                )))
+                .unwrap();
+            let writer_thread = std::thread::spawn(move || {
+                run_writer(&mut writer, write_rx);
+                writer
+            });
+            let input_write_tx = write_tx.clone();
+            let input_accepting = Arc::clone(&accepting);
+            let input_thread = std::thread::spawn(move || {
+                run_input_forwarder(&mut data_rx, input_write_tx, input_accepting)
+            });
+            flushed_rx.recv().expect("prompt was flushed");
+            during_delay(&write_tx, &accepting);
+            let result = reply_rx.recv().expect("writer reports submission");
+            drop(data_tx);
+            input_thread.join().expect("input thread joins");
+            drop(write_tx);
+            (writer_thread.join().expect("writer thread joins"), result)
+        }
+
+        #[test]
+        fn submission_sequences_user_input_but_allows_terminal_responses() {
+            let delay = Duration::from_millis(30);
+            let (writer, result) = run_recorded_submission(None, delay, |write_tx, _| {
+                write_tx
+                    .send(PtyIoWriteCommand::Write(Bytes::from_static(b"response")))
+                    .unwrap();
+            });
+            result.expect("submission succeeds");
+
+            assert_eq!(writer.writes[0].0, b"prompt");
+            assert_eq!(writer.writes[1].0, b"response");
+            assert_eq!(writer.writes[2].0, b"\r");
+            assert_eq!(writer.writes[3].0, b"user");
+            assert!(writer.writes[2].1.duration_since(writer.flushes[0]) >= delay);
+        }
+
+        #[test]
+        fn submission_returns_enter_write_failure() {
+            let (_writer, result) = run_recorded_submission(Some(1), Duration::ZERO, |_, _| {});
+            let err = result.expect_err("enter failure reaches caller");
+
+            assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
+        }
+
+        #[test]
+        fn shutdown_during_submission_delay_cancels_enter() {
+            let (writer, result) =
+                run_recorded_submission(None, Duration::from_millis(30), |_, accepting| {
+                    *accepting.lock().unwrap() = false;
+                });
+            let err = result.expect_err("shutdown cancels enter");
+            assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
+            assert_eq!(
+                writer
+                    .writes
+                    .iter()
+                    .map(|write| write.0.as_slice())
+                    .collect::<Vec<_>>(),
+                vec![b"prompt"]
+            );
+        }
+    }
 }

 #[cfg(windows)]
@@ -73,6 +73,12 @@ pub(crate) struct PtyIoActorConfig {

 enum PtyIoDataCommand {
    WriteUserInput(Bytes),
+    SubmitUserInput {
+        text: Bytes,
+        enter: Bytes,
+        delay: Duration,
+        reply: std_mpsc::Sender<std::io::Result<()>>,
+    },
 }

 enum PtyIoControlCommand {
@@ -100,37 +106,6 @@ struct UserWriteGate {
 }

 impl PtyIoActorHandle {
-    pub(crate) async fn write_user_input(
-        &self,
-        bytes: Bytes,
-    ) -> Result<(), mpsc::error::SendError<Bytes>> {
-        {
-            let user_writes = self
-                .user_writes
-                .lock()
-                .unwrap_or_else(|poisoned| poisoned.into_inner());
-            if !user_writes.accepting {
-                return Err(mpsc::error::SendError(bytes));
-            }
-        }
-
-        let permit = match self.data_tx.reserve().await {
-            Ok(permit) => permit,
-            Err(_) => return Err(mpsc::error::SendError(bytes)),
-        };
-
-        let user_writes = self
-            .user_writes
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner());
-        if !user_writes.accepting {
-            return Err(mpsc::error::SendError(bytes));
-        }
-        permit.send(PtyIoDataCommand::WriteUserInput(bytes));
-        self.wake_actor();
-        Ok(())
-    }
-
    pub(crate) fn try_write_user_input(
        &self,
        bytes: Bytes,
@@ -150,15 +125,57 @@ impl PtyIoActorHandle {
                self.wake_actor();
                Ok(())
            }
-            Err(mpsc::error::TrySendError::Full(PtyIoDataCommand::WriteUserInput(bytes))) => {
+            Err(mpsc::error::TrySendError::Full(command)) => {
+                let PtyIoDataCommand::WriteUserInput(bytes) = command else {
+                    unreachable!("queued write returned another command")
+                };
                Err(mpsc::error::TrySendError::Full(bytes))
            }
-            Err(mpsc::error::TrySendError::Closed(PtyIoDataCommand::WriteUserInput(bytes))) => {
+            Err(mpsc::error::TrySendError::Closed(command)) => {
+                let PtyIoDataCommand::WriteUserInput(bytes) = command else {
+                    unreachable!("queued write returned another command")
+                };
                Err(mpsc::error::TrySendError::Closed(bytes))
            }
        }
    }

+    pub(crate) fn queue_user_input_submission(
+        &self,
+        text: Bytes,
+        enter: Bytes,
+        delay: Duration,
+    ) -> std::io::Result<std_mpsc::Receiver<std::io::Result<()>>> {
+        let user_writes = self
+            .user_writes
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if !user_writes.accepting {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::BrokenPipe,
+                "pty actor closed",
+            ));
+        }
+        let (reply_tx, reply_rx) = std_mpsc::channel();
+        self.data_tx
+            .try_send(PtyIoDataCommand::SubmitUserInput {
+                text,
+                enter,
+                delay,
+                reply: reply_tx,
+            })
+            .map_err(|err| match err {
+                mpsc::error::TrySendError::Full(_) => {
+                    std::io::Error::new(std::io::ErrorKind::WouldBlock, "pty input queue is full")
+                }
+                mpsc::error::TrySendError::Closed(_) => {
+                    std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed")
+                }
+            })?;
+        self.wake_actor();
+        Ok(reply_rx)
+    }
+
    pub(crate) fn write_terminal_response(&self, response: impl FnOnce() -> Option<Bytes>) {
        let _order = self
            .response_order
@@ -232,6 +249,12 @@ impl PtyIoActorHandle {
                .user_writes
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            if !user_writes.accepting {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::WouldBlock,
+                    "PTY handoff is already in progress",
+                ));
+            }
            user_writes.accepting = false;
            if self
                .control_tx
@@ -391,6 +414,8 @@ impl PtyIoActor {
            },
            pending_writes: VecDeque::new(),
            current_write_offset: 0,
+            active_submission: None,
+            pending_handoff: None,
            wake_read_fd: wake_pipe.read_fd,
            controls,
            response_order,
@@ -421,8 +446,10 @@ struct PtyIoActorRunner {
    data_rx: mpsc::Receiver<PtyIoDataCommand>,
    control_rx: std_mpsc::Receiver<PtyIoControlCommand>,
    state: ActorState,
-    pending_writes: VecDeque<Bytes>,
+    pending_writes: VecDeque<PendingWrite>,
    current_write_offset: usize,
+    active_submission: Option<ActiveSubmission>,
+    pending_handoff: Option<std_mpsc::Sender<std::io::Result<()>>>,
    wake_read_fd: OwnedFd,
    controls: Arc<Mutex<SharedPtyControls>>,
    response_order: Arc<Mutex<()>>,
@@ -431,10 +458,47 @@ struct PtyIoActorRunner {
    poll_observer: Option<std_mpsc::Sender<()>>,
 }

+struct ActiveSubmission {
+    enter: Bytes,
+    delay: Duration,
+    phase: SubmissionPhase,
+    reply: std_mpsc::Sender<std::io::Result<()>>,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct PendingWrite {
+    bytes: Bytes,
+    boundary: Option<SubmissionBoundary>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum SubmissionBoundary {
+    Text,
+    Enter,
+}
+
+enum SubmissionPhase {
+    WritingText,
+    WaitingUntil(Instant),
+    WritingEnter,
+}
+
 impl PtyIoActorRunner {
    fn enqueue_write(&mut self, bytes: Bytes) {
        if !bytes.is_empty() {
-            self.pending_writes.push_back(bytes);
+            self.pending_writes.push_back(PendingWrite {
+                bytes,
+                boundary: None,
+            });
+        }
+    }
+
+    fn enqueue_submission_write(&mut self, bytes: Bytes, boundary: SubmissionBoundary) {
+        if !bytes.is_empty() {
+            self.pending_writes.push_back(PendingWrite {
+                bytes,
+                boundary: Some(boundary),
+            });
        }
    }

@@ -449,7 +513,18 @@ impl PtyIoActorRunner {
            self.apply_pending_controls();

            if !self.pending_writes.is_empty() {
-                self.flush_pending_writes_once();
+                match self.flush_pending_writes_once() {
+                    Ok(Some(boundary)) => self.complete_submission_boundary(boundary),
+                    Ok(None) => {}
+                    Err(err) => {
+                        self.fail_active_submission(err);
+                        break;
+                    }
+                }
+            }
+            self.schedule_submission_enter();
+            if self.active_submission.is_none() && self.pending_handoff.is_some() {
+                continue;
            }

            if let Some(poll_observer) = &self.poll_observer {
@@ -461,7 +536,7 @@ impl PtyIoActorRunner {
                self.wake_read_fd.as_raw_fd(),
                self.state == ActorState::Running,
                !self.pending_writes.is_empty(),
-                ACTOR_IDLE_POLL_MS,
+                self.poll_timeout_ms(),
            ) {
                Ok(readiness) => {
                    if readiness.wake_ready {
@@ -478,7 +553,14 @@ impl PtyIoActorRunner {
                        break;
                    }
                    if readiness.pty_write_ready && !self.pending_writes.is_empty() {
-                        self.flush_pending_writes_once();
+                        match self.flush_pending_writes_once() {
+                            Ok(Some(boundary)) => self.complete_submission_boundary(boundary),
+                            Ok(None) => {}
+                            Err(err) => {
+                                self.fail_active_submission(err);
+                                break;
+                            }
+                        }
                    }
                }
                Err(err) => {
@@ -488,6 +570,10 @@ impl PtyIoActorRunner {
            }
        }

+        self.fail_active_submission(std::io::Error::new(
+            std::io::ErrorKind::BrokenPipe,
+            "PTY actor closed during input submission",
+        ));
        if let Some(on_reader_exit) = self.on_reader_exit.take() {
            on_reader_exit();
        }
@@ -498,6 +584,13 @@ impl PtyIoActorRunner {
        if self.drain_control_commands() {
            return true;
        }
+        if self.active_submission.is_some() {
+            return false;
+        }
+        if let Some(reply) = self.pending_handoff.take() {
+            self.defer_or_begin_handoff(reply);
+            return false;
+        }
        self.drain_data_commands()
    }

@@ -530,6 +623,9 @@ impl PtyIoActorRunner {
                        should_exit = true;
                        break;
                    }
+                    if self.active_submission.is_some() {
+                        break;
+                    }
                }
                Err(DataTryRecvError::Empty) => break,
                Err(DataTryRecvError::Disconnected) => {
@@ -548,6 +644,32 @@ impl PtyIoActorRunner {
                    self.enqueue_write(bytes);
                }
            }
+            PtyIoDataCommand::SubmitUserInput {
+                text,
+                enter,
+                delay,
+                reply,
+            } => {
+                if self.state == ActorState::Running {
+                    let phase = if text.is_empty() {
+                        SubmissionPhase::WaitingUntil(Instant::now() + delay)
+                    } else {
+                        self.enqueue_submission_write(text, SubmissionBoundary::Text);
+                        SubmissionPhase::WritingText
+                    };
+                    self.active_submission = Some(ActiveSubmission {
+                        enter,
+                        delay,
+                        phase,
+                        reply,
+                    });
+                } else {
+                    let _ = reply.send(Err(std::io::Error::new(
+                        std::io::ErrorKind::BrokenPipe,
+                        "pty actor is not accepting input",
+                    )));
+                }
+            }
        }
        false
    }
@@ -555,8 +677,7 @@ impl PtyIoActorRunner {
    fn handle_control_command(&mut self, command: PtyIoControlCommand) -> bool {
        match command {
            PtyIoControlCommand::BeginHandoff(reply) => {
-                let result = self.begin_handoff();
-                let _ = reply.send(result);
+                self.defer_or_begin_handoff(reply);
            }
            PtyIoControlCommand::DuplicateForHandoff(reply) => {
                let result = if self.state == ActorState::Quiesced {
@@ -574,6 +695,7 @@ impl PtyIoActorRunner {
                let _ = reply.send(result);
            }
            PtyIoControlCommand::RollbackHandoff(reply) => {
+                self.pending_handoff.take();
                let result = if self.state == ActorState::Released {
                    Err(std::io::Error::new(
                        std::io::ErrorKind::BrokenPipe,
@@ -596,8 +718,26 @@ impl PtyIoActorRunner {
        false
    }

+    fn defer_or_begin_handoff(&mut self, reply: std_mpsc::Sender<std::io::Result<()>>) {
+        if self.active_submission.is_none() {
+            self.drain_pre_quiesce_commands();
+        }
+        if self.active_submission.is_some() {
+            self.pending_handoff = Some(reply);
+        } else {
+            let result = self.begin_handoff();
+            let _ = reply.send(result);
+        }
+    }
+
    fn begin_handoff(&mut self) -> std::io::Result<()> {
        self.drain_pre_quiesce_commands();
+        if self.active_submission.is_some() {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::WouldBlock,
+                "PTY input submission is still in progress",
+            ));
+        }
        self.apply_pending_controls();
        if self.state == ActorState::Released {
            return Err(std::io::Error::new(
@@ -606,7 +746,7 @@ impl PtyIoActorRunner {
            ));
        }
        let deadline = Instant::now() + HANDOFF_DRAIN_TIMEOUT;
-        self.flush_pending_writes_once();
+        let _ = self.flush_pending_writes_once()?;
        while !self.pending_writes.is_empty() {
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
@@ -633,7 +773,7 @@ impl PtyIoActorRunner {
                ));
            }
            if readiness.pty_write_ready {
-                self.flush_pending_writes_once();
+                let _ = self.flush_pending_writes_once()?;
            }
        }
        self.state = ActorState::Quiesced;
@@ -641,9 +781,12 @@ impl PtyIoActorRunner {
    }

    fn drain_pre_quiesce_commands(&mut self) {
-        while let Ok(PtyIoDataCommand::WriteUserInput(bytes)) = self.data_rx.try_recv() {
-            if self.state != ActorState::Released {
-                self.enqueue_write(bytes);
+        while let Ok(command) = self.data_rx.try_recv() {
+            if self.handle_data_command(command) {
+                break;
+            }
+            if self.active_submission.is_some() {
+                break;
            }
        }
    }
@@ -717,32 +860,100 @@ impl PtyIoActorRunner {
        }
    }

-    fn flush_pending_writes_once(&mut self) {
-        while let Some(bytes) = self.pending_writes.front() {
-            let chunk = &bytes[self.current_write_offset..];
+    fn complete_submission_boundary(&mut self, boundary: SubmissionBoundary) {
+        match boundary {
+            SubmissionBoundary::Text => {
+                let Some(submission) = self.active_submission.as_mut() else {
+                    return;
+                };
+                debug_assert!(matches!(submission.phase, SubmissionPhase::WritingText));
+                submission.phase = SubmissionPhase::WaitingUntil(Instant::now() + submission.delay);
+            }
+            SubmissionBoundary::Enter => {
+                let Some(submission) = self.active_submission.take() else {
+                    return;
+                };
+                debug_assert!(matches!(submission.phase, SubmissionPhase::WritingEnter));
+                let _ = submission.reply.send(Ok(()));
+            }
+        }
+    }
+
+    fn schedule_submission_enter(&mut self) {
+        let Some(ActiveSubmission {
+            enter,
+            phase: SubmissionPhase::WaitingUntil(deadline),
+            ..
+        }) = self.active_submission.as_ref()
+        else {
+            return;
+        };
+        if Instant::now() >= *deadline {
+            let enter = enter.clone();
+            if enter.is_empty() {
+                let submission = self.active_submission.take().unwrap();
+                let _ = submission.reply.send(Ok(()));
+            } else {
+                self.active_submission.as_mut().unwrap().phase = SubmissionPhase::WritingEnter;
+                self.enqueue_submission_write(enter, SubmissionBoundary::Enter);
+            }
+        }
+    }
+
+    fn poll_timeout_ms(&self) -> i32 {
+        let Some(ActiveSubmission {
+            phase: SubmissionPhase::WaitingUntil(deadline),
+            ..
+        }) = self.active_submission.as_ref()
+        else {
+            return ACTOR_IDLE_POLL_MS;
+        };
+        deadline
+            .saturating_duration_since(Instant::now())
+            .as_millis()
+            .max(1)
+            .min(ACTOR_IDLE_POLL_MS as u128) as i32
+    }
+
+    fn fail_active_submission(&mut self, err: std::io::Error) {
+        if let Some(submission) = self.active_submission.take() {
+            let _ = submission.reply.send(Err(err));
+        }
+    }
+
+    fn flush_pending_writes_once(&mut self) -> std::io::Result<Option<SubmissionBoundary>> {
+        while let Some(write) = self.pending_writes.front() {
+            let chunk = &write.bytes[self.current_write_offset..];
            match self.file.write(chunk) {
                Ok(0) => {
-                    warn!(pane = self.pane_id, "PTY actor write returned zero bytes");
-                    return;
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::WriteZero,
+                        "PTY actor write returned zero bytes",
+                    ));
                }
                Ok(written) => {
                    self.current_write_offset += written;
-                    if self.current_write_offset >= bytes.len() {
-                        self.pending_writes.pop_front();
+                    if self.current_write_offset >= write.bytes.len() {
+                        let completed = self.pending_writes.pop_front().unwrap();
                        self.current_write_offset = 0;
+                        if let Some(boundary) = completed.boundary {
+                            self.file.flush()?;
+                            return Ok(Some(boundary));
+                        }
                    }
                }
-                Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => return,
-                Err(err) if err.kind() == std::io::ErrorKind::Interrupted => return,
+                Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => return Ok(None),
+                Err(err) if err.kind() == std::io::ErrorKind::Interrupted => return Ok(None),
                Err(err) => {
                    warn!(pane = self.pane_id, err = %err, "PTY actor write failed");
                    self.pending_writes.clear();
                    self.current_write_offset = 0;
-                    return;
+                    return Err(err);
                }
            }
        }
-        let _ = self.file.flush();
+        self.file.flush()?;
+        Ok(None)
    }

    fn resize(&self, resize: PtyResize) {
@@ -879,6 +1090,8 @@ mod tests {
            state: ActorState::Running,
            pending_writes: VecDeque::new(),
            current_write_offset: 0,
+            active_submission: None,
+            pending_handoff: None,
            wake_read_fd: wake_pipe.read_fd,
            controls: Arc::new(Mutex::new(SharedPtyControls::default())),
            response_order: Arc::new(Mutex::new(())),
@@ -898,6 +1111,22 @@ mod tests {
        assert!(runner.pending_writes.is_empty());
    }

+    #[test]
+    fn submission_boundary_does_not_wait_for_following_protocol_write() {
+        let (mut runner, _peer) = actor_runner_for_unit_test();
+        runner.enqueue_submission_write(Bytes::from_static(b"prompt"), SubmissionBoundary::Text);
+        runner.enqueue_write(Bytes::from_static(b"response"));
+
+        assert_eq!(
+            runner.flush_pending_writes_once().unwrap(),
+            Some(SubmissionBoundary::Text)
+        );
+        assert_eq!(
+            runner.pending_writes[0].bytes,
+            Bytes::from_static(b"response")
+        );
+    }
+
    #[test]
    fn actor_writes_user_input_to_owned_fd() {
        let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
@@ -912,6 +1141,127 @@ mod tests {
        handle.shutdown();
    }

+    #[test]
+    fn actor_delays_enter_from_completed_prompt_write() {
+        let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
+        let text = Bytes::from(vec![b'x'; 4 * 1024 * 1024]);
+        let text_len = text.len();
+        let delay = Duration::from_millis(200);
+        let reader = std::thread::spawn(move || {
+            std::thread::sleep(delay);
+            let mut received = vec![0; text_len];
+            peer.read_exact(&mut received)
+                .expect("peer receives prompt");
+            let prompt_completed = Instant::now();
+            let mut enter = [0; 1];
+            peer.read_exact(&mut enter).expect("peer receives enter");
+            let enter_received = Instant::now();
+            let mut user = [0; 4];
+            peer.read_exact(&mut user)
+                .expect("peer receives queued input");
+            (prompt_completed, enter_received, enter, user)
+        });
+
+        let completion = handle
+            .queue_user_input_submission(text, Bytes::from_static(b"\r"), delay)
+            .expect("submission queues");
+        handle
+            .try_write_user_input(Bytes::from_static(b"user"))
+            .expect("ordinary input queues behind submission");
+        completion
+            .recv()
+            .expect("actor reports submission")
+            .expect("submission completes");
+        let (prompt_completed, enter_received, enter, user) = reader.join().expect("reader joins");
+
+        assert_eq!(enter, *b"\r");
+        assert_eq!(user, *b"user");
+        assert!(enter_received.duration_since(prompt_completed) >= delay / 2);
+
+        let err = match handle.queue_user_input_submission(
+            Bytes::from_static(b"prompt"),
+            Bytes::from_static(b"\r"),
+            Duration::ZERO,
+        ) {
+            Ok(completion) => completion
+                .recv()
+                .expect("actor reports submission")
+                .expect_err("closed PTY rejects submission"),
+            Err(err) => err,
+        };
+
+        assert!(matches!(
+            err.kind(),
+            std::io::ErrorKind::BrokenPipe
+                | std::io::ErrorKind::ConnectionReset
+                | std::io::ErrorKind::WriteZero
+        ));
+    }
+
+    #[test]
+    fn actor_completes_empty_submission_parts() {
+        let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
+        peer.set_read_timeout(Some(Duration::from_secs(1)))
+            .expect("peer timeout");
+
+        let completion = handle
+            .queue_user_input_submission(Bytes::new(), Bytes::from_static(b"\r"), Duration::ZERO)
+            .expect("empty prompt submission queues");
+        let mut enter = [0; 1];
+        peer.read_exact(&mut enter)
+            .expect("peer receives enter for empty prompt");
+        assert_eq!(enter, *b"\r");
+        completion
+            .recv_timeout(Duration::from_secs(1))
+            .expect("actor reports empty prompt submission")
+            .expect("empty prompt submission completes");
+
+        let completion = handle
+            .queue_user_input_submission(
+                Bytes::from_static(b"prompt"),
+                Bytes::new(),
+                Duration::from_millis(40),
+            )
+            .expect("empty enter submission queues");
+        let handoff_handle = handle.clone();
+        let handoff =
+            std::thread::spawn(move || handoff_handle.begin_handoff(Duration::from_millis(250)));
+        let mut prompt = [0; 6];
+        peer.read_exact(&mut prompt)
+            .expect("peer receives prompt before empty enter");
+        assert_eq!(&prompt, b"prompt");
+        completion
+            .recv_timeout(Duration::from_secs(1))
+            .expect("actor reports empty enter submission")
+            .expect("empty enter submission completes");
+        handoff
+            .join()
+            .expect("handoff thread joins")
+            .expect("handoff resumes without an idle poll after submission");
+        handle.shutdown();
+    }
+
+    #[test]
+    fn actor_reports_peer_closure_during_submission_delay() {
+        let (handle, mut peer, _read_rx) = actor_with_socket_pair(false);
+        let completion = handle
+            .queue_user_input_submission(
+                Bytes::from_static(b"prompt"),
+                Bytes::from_static(b"\r"),
+                Duration::from_secs(1),
+            )
+            .expect("submission queues");
+        let mut prompt = [0; 6];
+        peer.read_exact(&mut prompt).expect("peer receives prompt");
+        drop(peer);
+
+        let err = completion
+            .recv_timeout(Duration::from_secs(1))
+            .expect("actor reports peer closure")
+            .expect_err("peer closure fails the active submission");
+        assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
+    }
+
    #[test]
    fn actor_wakes_idle_poll_for_user_input() {
        let (poll_tx, poll_rx) = std_mpsc::channel();
@@ -976,9 +1326,9 @@ mod tests {
        .expect("actor spawn");

        let marker = Bytes::from_static(b"queued-input");
-        handle
-            .try_write_user_input(marker.clone())
-            .expect("write command accepted");
+        let completion = handle
+            .queue_user_input_submission(marker.clone(), Bytes::from_static(b"\r"), Duration::ZERO)
+            .expect("submission accepted");

        const OUTPUT_LEN: usize = 128 * 1024;
        let mut peer_writer = peer.try_clone().expect("clone peer writer");
@@ -1004,11 +1354,27 @@ mod tests {
        assert_eq!(output_len, OUTPUT_LEN);
        output_writer.join().expect("output writer joins");

-        let mut received_input = vec![0; prefilled + marker.len()];
+        let handoff_handle = handle.clone();
+        let handoff =
+            std::thread::spawn(move || handoff_handle.begin_handoff(Duration::from_secs(1)));
+
+        let mut received_input = vec![0; prefilled + marker.len() + 1];
        peer.read_exact(&mut received_input)
            .expect("peer receives prefill and queued input");
        assert!(received_input[..prefilled].iter().all(|byte| *byte == 0xAA));
-        assert_eq!(&received_input[prefilled..], marker.as_ref());
+        assert_eq!(
+            &received_input[prefilled..prefilled + marker.len()],
+            marker.as_ref()
+        );
+        assert_eq!(received_input.last(), Some(&b'\r'));
+        completion
+            .recv_timeout(Duration::from_secs(1))
+            .expect("actor reports submission")
+            .expect("submission completes");
+        handoff
+            .join()
+            .expect("handoff thread joins")
+            .expect("handoff waits for submission");
        handle.shutdown();
    }

@@ -1080,6 +1446,10 @@ mod tests {
        handle
            .begin_handoff(Duration::from_secs(1))
            .expect("handoff quiesced");
+        let err = handle
+            .begin_handoff(Duration::from_secs(1))
+            .expect_err("concurrent handoff rejected");
+        assert_eq!(err.kind(), std::io::ErrorKind::WouldBlock);
        assert!(handle
            .try_write_user_input(Bytes::from_static(b"blocked"))
            .is_err());
@@ -1205,6 +1575,8 @@ mod tests {
            state: ActorState::Running,
            pending_writes: VecDeque::new(),
            current_write_offset: 0,
+            active_submission: None,
+            pending_handoff: None,
            wake_read_fd: wake_pipe.read_fd,
            controls: Arc::clone(&controls),
            response_order: Arc::clone(&response_order),
@@ -1251,8 +1623,14 @@ mod tests {
        assert_eq!(
            runner.pending_writes,
            VecDeque::from([
-                Bytes::from_static(b"live-light"),
-                Bytes::from_static(b"query-light"),
+                PendingWrite {
+                    bytes: Bytes::from_static(b"live-light"),
+                    boundary: None,
+                },
+                PendingWrite {
+                    bytes: Bytes::from_static(b"query-light"),
+                    boundary: None,
+                },
            ])
        );
    }
@@ -1271,107 +1649,6 @@ mod tests {
        handle.shutdown();
    }

-    #[tokio::test]
-    async fn async_user_input_waits_for_queue_capacity() {
-        let (data_tx, mut data_rx) = mpsc::channel(1);
-        let (control_tx, _control_rx) = std_mpsc::channel();
-        data_tx
-            .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
-                b"fill",
-            )))
-            .expect("fill data queue");
-        let (wake, _wake_read_fd) = test_wake_pair();
-        let handle = PtyIoActorHandle {
-            data_tx,
-            control_tx,
-            wake,
-            user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
-            controls: Arc::new(Mutex::new(SharedPtyControls::default())),
-            response_order: Arc::new(Mutex::new(())),
-        };
-
-        let write = tokio::spawn(async move {
-            handle
-                .write_user_input(Bytes::from_static(b"wait-for-capacity"))
-                .await
-        });
-        tokio::time::sleep(Duration::from_millis(50)).await;
-        assert!(
-            !write.is_finished(),
-            "async input should wait for queue capacity"
-        );
-
-        assert!(matches!(
-            data_rx.recv().await,
-            Some(PtyIoDataCommand::WriteUserInput(_))
-        ));
-        write
-            .await
-            .expect("write task joins")
-            .expect("write succeeds after capacity opens");
-        match data_rx.recv().await {
-            Some(PtyIoDataCommand::WriteUserInput(bytes)) => {
-                assert_eq!(bytes, Bytes::from_static(b"wait-for-capacity"));
-            }
-            _ => panic!("expected queued user input"),
-        }
-    }
-
-    #[tokio::test]
-    async fn async_user_input_waiting_for_capacity_is_rejected_after_handoff_begins() {
-        let (data_tx, mut data_rx) = mpsc::channel(1);
-        let (control_tx, control_rx) = std_mpsc::channel();
-        data_tx
-            .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static(
-                b"fill",
-            )))
-            .expect("fill data queue");
-        let (wake, _wake_read_fd) = test_wake_pair();
-        let handle = PtyIoActorHandle {
-            data_tx,
-            control_tx,
-            wake,
-            user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
-            controls: Arc::new(Mutex::new(SharedPtyControls::default())),
-            response_order: Arc::new(Mutex::new(())),
-        };
-        let write_handle = handle.clone();
-        let write = tokio::spawn(async move {
-            write_handle
-                .write_user_input(Bytes::from_static(b"after-handoff-start"))
-                .await
-        });
-        tokio::time::sleep(Duration::from_millis(50)).await;
-
-        let handoff = std::thread::spawn(move || handle.begin_handoff(Duration::from_secs(1)));
-        match control_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("handoff control command")
-        {
-            PtyIoControlCommand::BeginHandoff(reply) => {
-                reply.send(Ok(())).expect("handoff waiter alive");
-            }
-            _ => panic!("expected begin handoff command"),
-        }
-        handoff
-            .join()
-            .expect("handoff thread joins")
-            .expect("handoff succeeds");
-        assert!(matches!(
-            data_rx.recv().await,
-            Some(PtyIoDataCommand::WriteUserInput(_))
-        ));
-
-        let err = write.await.expect("write task joins").expect_err(
-            "write waiting for capacity must be rejected after handoff closes the input gate",
-        );
-        assert_eq!(err.0, Bytes::from_static(b"after-handoff-start"));
-        match tokio::time::timeout(Duration::from_millis(50), data_rx.recv()).await {
-            Err(_) | Ok(None) => {}
-            Ok(Some(_)) => panic!("rejected write must not be queued"),
-        }
-    }
-
    #[test]
    fn handoff_control_is_not_blocked_by_full_data_queue() {
        let (data_tx, _data_rx) = mpsc::channel(1);
@@ -1431,6 +1708,8 @@ mod tests {
            state: ActorState::Running,
            pending_writes: VecDeque::new(),
            current_write_offset: 0,
+            active_submission: None,
+            pending_handoff: None,
            wake_read_fd: fd::create_wake_pipe().expect("wake pipe").read_fd,
            controls: Arc::new(Mutex::new(SharedPtyControls::default())),
            response_order: Arc::new(Mutex::new(())),
@@ -3037,6 +3037,12 @@ impl HeadlessServer {
            return changed;
        }
        let alt_screen_read_spec = self.alt_screen_read_spec(&msg.request);
+        if matches!(&msg.request.method, api::schema::Method::AgentPrompt(_)) {
+            let deferred_changed = self
+                .app
+                .handle_deferred_agent_api_request(msg.request, msg.respond_to);
+            return changed | deferred_changed;
+        }
        if matches!(
            &msg.request.method,
            api::schema::Method::WorktreeCreate(_) | api::schema::Method::WorktreeRemove(_)
@@ -449,8 +449,13 @@ impl TerminalRuntime {
        self.0.try_send_bytes(bytes)
    }

-    pub fn send_bytes_after(&self, bytes: Bytes, delay: std::time::Duration) {
-        self.0.send_bytes_after(bytes, delay);
+    pub fn queue_user_input_submission(
+        &self,
+        text: Bytes,
+        enter: Bytes,
+        delay: std::time::Duration,
+    ) -> std::io::Result<std::sync::mpsc::Receiver<std::io::Result<()>>> {
+        self.0.queue_user_input_submission(text, enter, delay)
    }

    pub fn try_send_paste(&self, text: String) -> Result<(), mpsc::error::TrySendError<Bytes>> {
@@ -223,11 +223,9 @@ fn agent_start_command_works() {
    fs::write(
        &fake_pi,
        format!(
-            "#!/bin/sh\nprintf '%s\\n' \"$@\" > '{}'\nexport HERDR_AGENT=pi\n'{}' pane report-agent \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --state idle >/dev/null\nwhile IFS= read -r prompt; do\n  case \"$prompt\" in \"do not transition\"|\"stall\") continue ;; esac\n  '{}' pane report-agent \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --state working >/dev/null\n  '{}' pane report-agent \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --state idle >/dev/null\n  printf '%s\\n' \"$prompt\" >> '{}'\ndone\n",
+            "#!/bin/sh\nprintf '%s\\n' \"$@\" > '{0}'\nexport HERDR_AGENT=pi\n'{1}' pane report-agent \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --state idle >/dev/null\nwhile IFS= read -r prompt; do\n  case \"$prompt\" in\n    \"do not transition\") continue ;;\n    \"done churn\")\n      '{1}' pane report-agent \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --state done >/dev/null\n      '{1}' pane report-agent \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --state idle >/dev/null\n      continue\n      ;;\n    \"session churn\")\n      '{1}' pane report-agent-session \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --agent-session-id replacement >/dev/null\n      continue\n      ;;\n    \"block after submit\")\n      '{1}' pane report-agent \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --state blocked >/dev/null\n      continue\n      ;;\n  esac\n  '{1}' pane report-agent \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --state working >/dev/null\n  '{1}' pane report-agent \"$HERDR_PANE_ID\" --source custom:fake-pi --agent pi --state idle >/dev/null\n  printf '%s\\n' \"$prompt\" >> '{2}'\ndone\n",
            captured_args.display(),
            env!("CARGO_BIN_EXE_herdr"),
-            env!("CARGO_BIN_EXE_herdr"),
-            env!("CARGO_BIN_EXE_herdr"),
            captured_prompts.display(),
        ),
    )
@@ -394,69 +392,74 @@ fn agent_start_command_works() {
    thread::sleep(Duration::from_millis(400));
    assert_eq!(fs::read(&captured_prompts).unwrap(), prompts_before_blocked);

-    let idle_report = run_cli(
-        &socket_path,
-        &[
-            "pane",
-            "report-agent",
-            &pane_id,
-            "--source",
-            "custom:fake-pi",
-            "--agent",
-            "pi",
-            "--state",
-            "idle",
-        ],
-    );
-    assert!(idle_report.status.success());
+    let report_agent = |state| {
+        run_cli(
+            &socket_path,
+            &[
+                "pane",
+                "report-agent",
+                &pane_id,
+                "--source",
+                "custom:fake-pi",
+                "--agent",
+                "pi",
+                "--state",
+                state,
+            ],
+        )
+        .status
+        .success()
+    };
+    let prompt_wait = |prompt, timeout| {
+        run_cli(
+            &socket_path,
+            &[
+                "agent",
+                "prompt",
+                "main",
+                prompt,
+                "--wait",
+                "--timeout",
+                timeout,
+            ],
+        )
+    };

-    let stale_idle = run_cli(
-        &socket_path,
-        &[
-            "agent",
-            "prompt",
-            "main",
-            "do not transition",
-            "--wait",
-            "--timeout",
-            "500",
-        ],
-    );
+    assert!(report_agent("idle"));
+    let stale_idle = prompt_wait("do not transition", "500");
    assert_eq!(stale_idle.status.code(), Some(1));
    let stale_idle: serde_json::Value = serde_json::from_slice(&stale_idle.stderr).unwrap();
    assert_eq!(stale_idle["error"]["code"], "timeout");

-    let stalled = run_cli(
-        &socket_path,
-        &[
-            "agent",
-            "prompt",
-            "main",
-            "stall",
-            "--wait",
-            "--timeout",
-            "6000",
-        ],
-    );
+    let stalled = prompt_wait("do not transition", "6000");
    assert_eq!(stalled.status.code(), Some(1));
    let stalled: serde_json::Value = serde_json::from_slice(&stalled.stderr).unwrap();
    assert_eq!(stalled["error"]["code"], "agent_prompt_stalled");
    assert!(stalled["error"]["message"]
        .as_str()
-        .is_some_and(|message| message.contains("state_change_seq remained")));
+        .is_some_and(|message| message.contains("no observed working or blocked state")));

-    let prompted = run_cli(
-        &socket_path,
-        &[
-            "agent",
-            "prompt",
-            "main",
-            "Review this diff",
-            "--wait",
-            "--timeout",
-            "2000",
-        ],
+    for prompt in ["done churn", "session churn"] {
+        let settled_only = prompt_wait(prompt, "500");
+        assert_eq!(settled_only.status.code(), Some(1));
+        let settled_only: serde_json::Value = serde_json::from_slice(&settled_only.stderr).unwrap();
+        assert_eq!(settled_only["error"]["code"], "timeout");
+    }
+
+    let blocked_after_submit = prompt_wait("block after submit", "2000");
+    assert!(blocked_after_submit.status.success());
+    let blocked_after_submit: serde_json::Value =
+        serde_json::from_slice(&blocked_after_submit.stdout).unwrap();
+    assert_eq!(
+        blocked_after_submit["result"]["agent"]["agent_status"],
+        "blocked"
    );
+    assert!(report_agent("idle"));
+    assert!(report_agent("working"));
+    let already_working = prompt_wait("finish active", "2000");
+    assert!(already_working.status.success());
+
+    let prompted = prompt_wait("Review this diff", "2000");
    assert!(
        prompted.status.success(),
        "prompt failed: {}",