You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by zh...@apache.org on 2026/10/16 09:21:35 UTC
(maka) branch feat/cli-run-interaction created (now efb93decd7)
This is an automated email from the ASF dual-hosted git repository.
me2seeks pushed a change to branch feat/cli-run-interaction
in repository https://gitbox.apache.org/repos/asf/maka.git
at efb93decd7 feat(cli): add session-based run interaction
This branch includes the following new commits:
new efb93decd7 feat(cli): add session-based run interaction
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
(maka) 01/01: feat(cli): add session-based run interaction
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
me2seeks pushed a commit to branch feat/cli-run-interaction
in repository https://gitbox.apache.org/repos/asf/maka.git
commit efb93decd78f9c2467b1499a55c265b6612d07af
Author: me2seeks <me...@gmail.com>
AuthorDate: Fri Oct 16 17:21:13 2026 +0800
feat(cli): add session-based run interaction
Let supervising agents wait for a Session, receive and answer built-in
AskUser questions across CLI invocations, queue or steer input, and read
durable results. Preserve the existing blocking execution path and MCP
lifetime constraints. Guard admission against cancellation during reads.
Generated-by: Codex
---
packages/cli/README.md | 51 +++
packages/cli/README.zh-CN.md | 43 +++
packages/cli/src/__tests__/run-command.test.ts | 50 +++
.../cli/src/__tests__/run-session-host.test.ts | 280 +++++++++++++++
.../src/__tests__/run-session-process-fixture.ts | 26 ++
.../src/__tests__/runtime-host-run-command.test.ts | 76 +++-
packages/cli/src/run-command-core.ts | 171 ++++++++-
packages/cli/src/run-session-command.ts | 391 +++++++++++++++++++++
packages/cli/src/run-session-result.ts | 52 +++
packages/cli/src/run-turn-outcome.ts | 229 ++++++++++++
packages/cli/src/runtime-host-run-command.ts | 278 ++++-----------
packages/cli/src/runtime-host-run-mcp.ts | 4 +
12 files changed, 1428 insertions(+), 223 deletions(-)
diff --git a/packages/cli/README.md b/packages/cli/README.md
index 81865e454f..f6c8c01917 100644
--- a/packages/cli/README.md
+++ b/packages/cli/README.md
@@ -116,6 +116,57 @@ Maka asks before privileged tool operations by default. `maka run --yolo` grants
and network access and should only be used in an environment you are prepared to let the task
modify.
+
+For another agent to drive a Session, request structured output or explicitly submit without waiting:
+
+```sh
+maka run "Inspect this project" --json
+maka run "Inspect this project" --no-wait
+```
+
+Both forms return a `sessionId` in their JSON result. Ordinary runs still wait and print the final
+answer by default. `--no-wait` returns after admission; use the Session ID to observe the eventual
+result. Given a returned Session ID in `$session`:
+
+```sh
+maka run --session "$session" --status
+maka run --session "$session" --wait --json
+maka run --session "$session" --answer '["Yes",null]' --wait --json
+maka run --session "$session" "Inspect the tests next" --queue --wait --json
+maka run --session "$session" "Focus on the parser" --steer --no-wait
+maka run --session "$session" --result --json
+maka run --session "$session" --resume --wait --json
+```
+
+An ordinary built-in AskUser returns `needs_input`, the questions and options, with exit code 0.
+The Host retains the waiting Turn. `--answer` accepts one string or `null` per question in a JSON
+array and continues the original tool call and Turn. It does not need `--resume`. A failed or
+uncertain answer submission is not automatically retried; read the current questions before
+choosing what to do next. Multiple pending batches are reported, and an answer is refused when
+the current batch cannot be selected uniquely.
+
+Queue is the default: each explicit queued message opens its own subsequent Turn. Steer uses the
+Host's existing current-Turn delivery and may carry over if it misses that Turn's last boundary.
+Submitting a queued prompt with `--wait` follows that message's execution. Steer with `--wait`
+observes the Turn selected on attachment; it does not guarantee consumption before that Turn ends. A standalone `--wait` fixes its
+target on attachment; when idle, it reads the latest result. `--result` never waits for an active
+Turn. Status and result queries do not prepare a model or start MCP. Interrupting Session
+observation with Ctrl+C leaves Host execution running; legacy blocking run cancellation retains
+its existing behavior.
+
+Session commands use the saved configuration. Keep using the legacy `--resume <session-id>`
+selector for blocking runs that supply configuration, cwd, timeout or Graph options; that selector
+continues accepting a new prompt. `--session <id> --resume` instead requests safe continuation of
+stopped execution. An idle Session prompt can use `--max-steps`; a queued or steering message
+cannot attach a step cap to an existing Turn. Executing a full-access Session still requires
+`--yolo`.
+
+Returning from client-owned MCP interaction and resuming it in another CLI process is not
+supported. Blocking MCP runs retain their existing behavior; `--no-wait` refuses to detach the
+MCP servers owned by that invocation. Graph interaction handoff is also deferred. Use a persistent
+CLI installation for local Session control execution and detached runs; temporary npx invocations
+may own the Host's lifetime and keep the legacy blocking path.
+
## Upgrade
Update within the nightly line by pinning the exact release (`--target` accepts `latest`,
diff --git a/packages/cli/README.zh-CN.md b/packages/cli/README.zh-CN.md
index ff32c67b41..79e3a0cd6f 100644
--- a/packages/cli/README.zh-CN.md
+++ b/packages/cli/README.zh-CN.md
@@ -106,6 +106,49 @@ maka run --help
Maka 默认会在执行高权限工具操作前询问。`maka run --yolo` 会授予该任务完整的文件和网络
权限，只应在你允许任务修改的环境中使用。
+
+主代理可以请求结构化结果，或显式提交后返回：
+
+```sh
+maka run "检查这个项目" --json
+maka run "检查这个项目" --no-wait
+```
+
+这两种调用的 JSON 结果都包含 `sessionId`。普通调用仍默认等待并输出最终答案。
+`--no-wait` 在提交被接收后返回；使用会话 ID 查询后续结果。假设返回的会话 ID 保存在
+`$session` 中：
+
+```sh
+maka run --session "$session" --status
+maka run --session "$session" --wait --json
+maka run --session "$session" --answer '["是",null]' --wait --json
+maka run --session "$session" "接下来检查测试" --queue --wait --json
+maka run --session "$session" "重点检查参数解析" --steer --no-wait
+maka run --session "$session" --result --json
+maka run --session "$session" --resume --wait --json
+```
+
+普通内置 AskUser 会以退出码 0 返回 `needs_input`、问题及选项，Host 保留等待中的 Turn。
+`--answer` 使用 JSON 数组，为每个问题提供一个字符串或 `null`，回答后原工具调用和原 Turn
+继续执行，无需 `--resume`。答案提交失败或结果不确定时不会自动重发；先重新读取当前问题，
+再决定下一步。存在多批待回答问题时会如实报告，无法唯一选择当前批次时拒绝提交答案。
+
+默认使用 queue，每条显式排队消息对应一个独立后续 Turn。steer 沿用 Host 当前 Turn 的追加
+机制，错过最后处理边界时可能进入后续 Turn。提交排队消息时的 `--wait` 跟踪该消息对应的执行；steer 配合 `--wait` 观察连接时选定的
+Turn，不保证追加消息一定在该 Turn 结束前消费。
+单独 `--wait` 在连接时固定目标，空闲时读取最近结果。`--result` 不等待仍在执行的 Turn。
+状态和结果查询不准备模型、不启动 MCP。用 Ctrl+C 退出会话观察会让 Host 继续执行；旧阻塞
+run 的取消行为保持原样。
+
+会话控制使用已保存的配置。需要指定配置、cwd、timeout 或 Graph 选项的阻塞运行，继续使用
+兼容入口 `--resume <session-id>` 选择会话并提供新 prompt。`--session <id> --resume` 则表示
+安全恢复已停止的执行。空闲会话的新 prompt 可以使用 `--max-steps`，排队或 steer 消息不能
+给已有 Turn 附加步数上限。执行完整权限会话仍需显式 `--yolo`。
+
+当前不支持客户端 MCP 交互返回后由另一个 CLI 进程续接。阻塞 MCP 运行保留现有行为；
+`--no-wait` 会拒绝脱离本次调用持有的 MCP servers。Graph 交互交接也暂缓。执行本地会话控制
+及后台提交时使用持久 CLI 安装；临时 npx 调用可能持有 Host 生命周期，继续使用旧阻塞路径。
+
## 升级
在 nightly 线内更新时固定精确版本（`--target` 只接受 `latest`、`next` 或精确的 Maka
diff --git a/packages/cli/src/__tests__/run-command.test.ts b/packages/cli/src/__tests__/run-command.test.ts
index 9e3e96699c..dc9087d639 100644
--- a/packages/cli/src/__tests__/run-command.test.ts
+++ b/packages/cli/src/__tests__/run-command.test.ts
@@ -39,6 +39,56 @@ describe('maka run argument parsing', () => {
assert.equal(parseMakaRunArgs(['x', '--max-steps', '1.5']).kind, 'error');
});
+ test('parses session controls while preserving the legacy resume selector', () => {
+ const parsed = parseMakaRunArgs(['--session', 'session-1', '--resume', '--wait', '--json']);
+ assert.equal(parsed.kind, 'run');
+ if (parsed.kind !== 'run') return;
+ assert.equal(parsed.options.sessionId, 'session-1');
+ assert.equal(parsed.options.resumeExecution, true);
+ assert.equal(parsed.options.resumeId, undefined);
+ assert.deepEqual(parseMakaRunArgs(['--session', 'session-1', '--answer', '["Yes",null]']), {
+ kind: 'run',
+ options: { stdinPrompt: false, sessionId: 'session-1', answers: ['Yes', null] },
+ });
+ for (const args of [
+ ['--status'],
+ ['--session', 's', '--answer', 'yes'],
+ ['--session', 's', '--answer', '[1]'],
+ ['--session', 's', '--status', '--result'],
+ ['--session', 's', 'x', '--queue', '--steer'],
+ ['x', '--wait', '--no-wait'],
+ ['--session', 's', '--queue'],
+ ['--session', 's', '--resume', 'extra'],
+ ['x', '--graph', '--no-wait'],
+ ])
+ assert.equal(parseMakaRunArgs(args).kind, 'error', args.join(' '));
+ });
+
+ test('session queries bypass prompt/stdin and model execution preparation', async () => {
+ const stdout: string[] = [];
+ const exit = await runMakaTextCliCore(
+ ['--session', 's', '--status'],
+ {
+ listSessions: async () => {
+ throw new Error('must not list model-compatible Sessions');
+ },
+ createContext: async () => {
+ throw new Error('provider is out of quota');
+ },
+ runSession: async (options) => ({ sessionId: options.sessionId!, status: 'running' }),
+ },
+ {
+ stdinIsTTY: () => false,
+ readStdin: async () => {
+ throw new Error('must not read stdin');
+ },
+ writeStdout: (text) => stdout.push(text),
+ },
+ );
+ assert.equal(exit, 0);
+ assert.deepEqual(JSON.parse(stdout.join('')), { sessionId: 's', status: 'running' });
+ });
+
test('accepts only the explicit non-interactive sandbox bypass flag', () => {
assert.deepEqual(parseMakaRunArgs(['run tools', '--yolo']), {
kind: 'run',
diff --git a/packages/cli/src/__tests__/run-session-host.test.ts b/packages/cli/src/__tests__/run-session-host.test.ts
new file mode 100644
index 0000000000..46ff64c355
--- /dev/null
+++ b/packages/cli/src/__tests__/run-session-host.test.ts
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { spawn, type ChildProcess } from 'node:child_process';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { test } from 'node:test';
+import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority';
+import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores';
+import {
+ FakeBackend,
+ FAKE_ASK_USER_QUESTION_PROMPT,
+ FAKE_WAIT_FOR_STEERING_PROMPT,
+ FAKE_ERROR_PROMPT_PREFIX,
+} from '@maka/runtime/test-only/fake-backend';
+import { startExecutionRuntimeHostService } from '@maka/runtime-host/server';
+import { runMakaSessionCommand } from '../run-session-command.js';
+import { RunSessionInterrupted } from '../run-session-result.js';
+import { randomUUID } from 'node:crypto';
+import { connectRuntimeHost } from '@maka/runtime-host/client';
+import { RUNTIME_HOST_PROTOCOL_VERSION } from '@maka/runtime-host/protocol';
+
+const children = new Set<ChildProcess>();
+
+const fixturePath = fileURLToPath(new URL('./run-session-process-fixture.js', import.meta.url));
+
+test('real Host keeps AskUser alive across CLI processes and returns the original Turn result', {
+ timeout: 30_000,
+}, async () => {
+ const root = await mkdtemp(join(tmpdir(), 'maka-run-session-'));
+ const capability = await resolveStorageRoot({ path: root, kind: 'interactive' });
+ const owner = await tryAcquireInteractiveRootOwner(capability);
+ assert.ok(owner);
+ const stores = await openInteractiveExecutionStoresForWrite(owner.lease);
+ let sessionId: string;
+ try {
+ sessionId = (
+ await stores.sessionStore.create({
+ cwd: root,
+ llmConnectionId: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa',
+ llmConnectionSlug: 'fake',
+ model: 'fake-model',
+ permissionMode: 'ask',
+ })
+ ).id;
+ } finally {
+ await stores.sessionStore.close?.();
+ await owner.close();
+ }
+ const { createExecutionRuntimeHostComposition } = await import(
+ new URL('./execution-composition.js', import.meta.resolve('@maka/runtime-host/server')).href
+ );
+ const host = await startExecutionRuntimeHostService(
+ { rootPath: root },
+ {
+ createComposition: (context, options) =>
+ createExecutionRuntimeHostComposition(
+ context,
+ { ...options, bootstrapRuntimePolicy: false },
+ {
+ primaryBackendFactory: (backendContext: ConstructorParameters<typeof FakeBackend>[0]) =>
+ new FakeBackend(backendContext),
+ generateSessionTitle: async () => undefined,
+ },
+ ),
+ },
+ );
+ try {
+ const asked = await cli(root, '--session', sessionId, FAKE_ASK_USER_QUESTION_PROMPT, '--json');
+ assert.equal(asked.code, 0, asked.stderr);
+ const question = JSON.parse(asked.stdout);
+ assert.equal(question.status, 'needs_input');
+ assert.ok(question.questions.length > 0);
+ const connected = await connectRuntimeHost({
+ rootPath: root,
+ protocol: { min: RUNTIME_HOST_PROTOCOL_VERSION, max: RUNTIME_HOST_PROTOCOL_VERSION },
+ });
+ assert.equal(connected.kind, 'connected');
+ const probe = connected.connection;
+ const subscription = await probe.openSessionSubscription({
+ sessionId,
+ transcript: { kind: 'none' },
+ });
+ await subscription.ready();
+ const original = subscription.snapshot.rootTurn!;
+ assert.ok(original.status === 'running' || original.status === 'waiting_for_user');
+ const answers = question.questions.map(
+ (q: { options: Array<{ label: string }> }) => q.options[0]!.label,
+ );
+ const status = await cli(root, '--session', sessionId, '--status');
+ assert.equal(JSON.parse(status.stdout).status, 'needs_input');
+ const answered = await cli(
+ root,
+ '--session',
+ sessionId,
+ '--answer',
+ JSON.stringify(answers),
+ '--json',
+ );
+ assert.equal(answered.code, 0, answered.stderr);
+ assert.equal(JSON.parse(answered.stdout).status, 'completed');
+ assert.match(JSON.parse(answered.stdout).finalOutput, /Fake question answers:/);
+ const terminal = await probe.request('turn.query', { sessionId, turnId: original.turnId });
+ assert.equal(terminal.status, 'completed');
+ assert.equal(terminal.runId, original.runId);
+ const result = await cli(root, '--session', sessionId, '--result', '--json');
+ assert.equal(result.code, 0, result.stderr);
+ assert.deepEqual(JSON.parse(result.stdout), JSON.parse(answered.stdout));
+ const duplicate = await cli(root, '--session', sessionId, '--answer', JSON.stringify(answers));
+ assert.equal(duplicate.code, 1);
+ assert.match(duplicate.stderr, /found 0/);
+ // Ctrl+C while a read is in flight must not admit a new message afterward.
+ let interrupt: (() => void) | undefined;
+ let interruptedAdmissions = 0;
+ const interruptedConnection = new Proxy(probe, {
+ get(target, key) {
+ if (key === 'request')
+ return async (...args: Parameters<typeof probe.request>) => {
+ const value = await Reflect.apply(probe.request, probe, args);
+ if (args[0] === 'session.execution_boundary.query') interrupt?.();
+ if (args[0] === 'turn.message.submit') interruptedAdmissions++;
+ return value;
+ };
+ const value = Reflect.get(target, key, target);
+ return typeof value === 'function' ? value.bind(target) : value;
+ },
+ });
+ await assert.rejects(
+ () =>
+ runMakaSessionCommand(
+ interruptedConnection,
+ {
+ sessionId,
+ stdinPrompt: false,
+ prompt: 'must-not-be-admitted',
+ noWait: true,
+ },
+ {
+ onSigint: (handler) => {
+ interrupt = handler;
+ return () => {
+ interrupt = undefined;
+ };
+ },
+ newId: randomUUID,
+ },
+ ),
+ RunSessionInterrupted,
+ );
+ assert.equal(interruptedAdmissions, 0);
+ // Keep the next Turn parked while two independent CLI messages are admitted.
+ const askedAgain = await cli(
+ root,
+ '--session',
+ sessionId,
+ FAKE_ASK_USER_QUESTION_PROMPT,
+ '--json',
+ );
+ assert.equal(askedAgain.code, 0, askedAgain.stderr);
+ const queuedFirst = await cli(root, '--session', sessionId, 'queued-first', '--no-wait');
+ assert.equal(queuedFirst.code, 0, queuedFirst.stderr);
+ const queuedSecond = cli(root, '--session', sessionId, 'queued-second', '--wait', '--json');
+ let admitted = false;
+ for (let attempt = 0; attempt < 10; attempt++) {
+ const current = await cli(root, '--session', sessionId, '--status');
+ if (JSON.parse(current.stdout).queued >= 2) {
+ admitted = true;
+ break;
+ }
+ }
+ assert.equal(admitted, true, 'both explicit follow-ups must be queued before answering');
+ const waitingCurrent = await cli(root, '--session', sessionId, '--wait');
+ assert.equal(JSON.parse(waitingCurrent.stdout).status, 'needs_input');
+ const answeredAgain = await cli(
+ root,
+ '--session',
+ sessionId,
+ '--answer',
+ JSON.stringify(answers),
+ '--json',
+ );
+ assert.equal(answeredAgain.code, 0, answeredAgain.stderr);
+ assert.match(JSON.parse(answeredAgain.stdout).finalOutput, /Fake question answers:/);
+ const secondResult = await queuedSecond;
+ assert.equal(secondResult.code, 0, secondResult.stderr);
+ assert.match(JSON.parse(secondResult.stdout).finalOutput, /queued-second/);
+ const turns = await probe.request('session.turns.query', {
+ sessionId,
+ throughSequence: null,
+ position: 0,
+ maxContributions: 128,
+ });
+ const first = turns.contributions.find((t) => t.userPromptPreview === 'queued-first');
+ const second = turns.contributions.find((t) => t.userPromptPreview === 'queued-second');
+ assert.ok(first);
+ assert.ok(second);
+ assert.notEqual(first.turnId, second.turnId);
+ const awaitingSteer = await cli(
+ root,
+ '--session',
+ sessionId,
+ FAKE_WAIT_FOR_STEERING_PROMPT,
+ '--no-wait',
+ );
+ assert.equal(awaitingSteer.code, 0, awaitingSteer.stderr);
+ const steered = await cli(
+ root,
+ '--session',
+ sessionId,
+ 'steer-this-turn',
+ '--steer',
+ '--wait',
+ '--json',
+ );
+ assert.equal(steered.code, 0, steered.stderr);
+ assert.match(JSON.parse(steered.stdout).finalOutput, /Acknowledged steering: steer-this-turn/);
+ const failed = await cli(
+ root,
+ '--session',
+ sessionId,
+ `${FAKE_ERROR_PROMPT_PREFIX}rate_limit`,
+ '--json',
+ );
+ assert.equal(failed.code, 1);
+ assert.match(JSON.parse(failed.stdout).error, /Deterministic rate_limit failure/);
+ const failedResult = await cli(root, '--session', sessionId, '--result', '--json');
+ assert.equal(failedResult.code, 1);
+ assert.match(JSON.parse(failedResult.stdout).error, /Deterministic rate_limit failure/);
+ await subscription.close();
+ await probe.close();
+ } finally {
+ for (const child of children) child.kill('SIGTERM');
+ await host.close();
+ await rm(root, { recursive: true, force: true });
+ }
+});
+
+async function cli(
+ root: string,
+ ...args: string[]
+): Promise<{ code: number | null; stdout: string; stderr: string }> {
+ const child = spawn(process.execPath, [fixturePath, root, ...args], {
+ stdio: ['ignore', 'pipe', 'pipe'],
+ });
+ children.add(child);
+ let stdout = '';
+ let stderr = '';
+ child.stdout!.setEncoding('utf8').on('data', (data) => {
+ stdout += data;
+ });
+ child.stderr!.setEncoding('utf8').on('data', (data) => {
+ stderr += data;
+ });
+ return await new Promise((resolve, reject) => {
+ child.on('error', reject);
+ child.on('close', (code) => {
+ children.delete(child);
+ resolve({ code, stdout, stderr });
+ });
+ });
+}
diff --git a/packages/cli/src/__tests__/run-session-process-fixture.ts b/packages/cli/src/__tests__/run-session-process-fixture.ts
new file mode 100644
index 0000000000..3050dd1349
--- /dev/null
+++ b/packages/cli/src/__tests__/run-session-process-fixture.ts
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { runRuntimeHostTextCli } from '../runtime-host-run-command.js';
+const [root, ...args] = process.argv.slice(2);
+process.exitCode = await runRuntimeHostTextCli(args, {
+ workspaceRoot: () => root!,
+ processCwd: () => root!,
+ stdinIsTTY: () => true,
+});
diff --git a/packages/cli/src/__tests__/runtime-host-run-command.test.ts b/packages/cli/src/__tests__/runtime-host-run-command.test.ts
index 4ad42390d6..d17d48a12f 100644
--- a/packages/cli/src/__tests__/runtime-host-run-command.test.ts
+++ b/packages/cli/src/__tests__/runtime-host-run-command.test.ts
@@ -35,6 +35,7 @@ import {
type SubscriptionFrame,
} from '@maka/runtime-host/protocol';
import { resolveRuntimeHostCliTarget } from '../runtime-host-cli-context.js';
+import { RunQuestionAvailable } from '../run-session-result.js';
import { createRuntimeHostRunContext, runRuntimeHostTextCli } from '../runtime-host-run-command.js';
import type { RuntimeHostMakaSessionDriver } from '../runtime-host-session-driver.js';
import type { MakaRunContextInput, MakaRunOutcome } from '../run-command-core.js';
@@ -548,7 +549,7 @@ describe('Runtime Host maka run adapter', () => {
);
assert.equal(exitCode, 1);
- assert.equal(stderr.join(''), 'maka run: Agent Graph final Turn failed\n');
+ assert.equal(stderr.join(''), 'maka run: Turn failed\n');
});
test('waits for Host-started graph supervisor Turns before returning', async () => {
@@ -903,7 +904,7 @@ describe('Runtime Host maka run adapter', () => {
assert.deepEqual(fixture.graphStops, [session.id, session.id]);
});
- test('fails and stops instead of waiting for an interactive question', async () => {
+ test('returns an AskUser handoff without stopping its Host Turn', async () => {
const fixture = runFixture({
turnEvents: questionEvents('turn-1'),
pendingInteractions: [pendingQuestion('turn-1')],
@@ -923,11 +924,74 @@ describe('Runtime Host maka run adapter', () => {
text: 'ask me something',
}),
),
- new Error('interactive user questions are unavailable in non-interactive mode'),
+ RunQuestionAvailable,
);
- assert.deepEqual(fixture.exactTurnStops, [
- { sessionId: session.id, turnId: 'turn-1', runId: 'run-1' },
- ]);
+ assert.deepEqual(fixture.exactTurnStops, []);
+ });
+
+ test('public run returns questions with exit 0 and keeps the original Turn waiting', async () => {
+ const fixture = runFixture({
+ turnEvents: questionEvents('turn-1'),
+ pendingInteractions: [pendingQuestion('turn-1')],
+ pendingAfterTurnStarts: true,
+ });
+ const stdout: string[] = [];
+ const stderr: string[] = [];
+ const code = await runFixtureCommand(
+ fixture,
+ ['ask something'],
+ (text) => stdout.push(text),
+ (text) => stderr.push(text),
+ );
+ assert.equal(code, 0, stderr.join(''));
+ assert.equal(JSON.parse(stdout.join('')).status, 'needs_input');
+ assert.equal(JSON.parse(stdout.join('')).sessionId, 'session-created');
+ assert.ok(JSON.parse(stdout.join('')).questions.length);
+ assert.deepEqual(fixture.exactTurnStops, []);
+ assert.deepEqual(stderr, []);
+ });
+
+ test('no-wait returns after admission instead of consuming the event stream', async () => {
+ const fixture = runFixture({
+ turnEvents: (async function* (): AsyncIterable<SessionEvent> {
+ throw new Error('no-wait must not consume events');
+ })(),
+ });
+ const stdout: string[] = [];
+ const stderr: string[] = [];
+ const code = await runFixtureCommand(
+ fixture,
+ ['execute in background', '--no-wait'],
+ (text) => stdout.push(text),
+ (text) => stderr.push(text),
+ );
+ assert.equal(code, 0, stderr.join(''));
+ assert.deepEqual(JSON.parse(stdout.join('')), {
+ sessionId: 'session-created',
+ status: 'running',
+ });
+ assert.deepEqual(fixture.exactTurnStops, []);
+ });
+
+ test('preserves the concrete provider failure stored with a Turn', async () => {
+ const messages = structuredClone(graphMessages());
+ const terminal = messages.find(
+ (message) => message.type === 'turn_state' && message.turnId === 'turn-2',
+ );
+ assert.ok(terminal && terminal.type === 'turn_state');
+ terminal.status = 'failed';
+ terminal.errorClass = 'quota_exhausted';
+ terminal.failureMessage = 'Selected provider has no remaining quota';
+ const fixture = runFixture({ graph: true, finalMessages: messages });
+ const stderr: string[] = [];
+ const code = await runFixtureCommand(
+ fixture,
+ ['execute', '--graph'],
+ () => {},
+ (text) => stderr.push(text),
+ );
+ assert.equal(code, 1);
+ assert.match(stderr.join(''), /Selected provider has no remaining quota/);
});
test('fails and stops instead of dropping an interactive form', async () => {
diff --git a/packages/cli/src/run-command-core.ts b/packages/cli/src/run-command-core.ts
index 22ed89ab1c..02b4a96ebb 100644
--- a/packages/cli/src/run-command-core.ts
+++ b/packages/cli/src/run-command-core.ts
@@ -17,6 +17,11 @@
* under the License.
*/
+import {
+ RunQuestionAvailable,
+ RunSessionInterrupted,
+ type MakaRunSessionResult,
+} from './run-session-result.js';
import { randomUUID } from 'node:crypto';
import { realpath, stat } from 'node:fs/promises';
import { resolve } from 'node:path';
@@ -32,6 +37,15 @@ import { sessionEventSandboxBoundaryFailureReason } from './sandbox-boundary-fai
import { resolveMakaWorkspaceRoot } from './workspace-root.js';
export interface MakaRunOptions {
+ sessionId?: string;
+ wait?: true;
+ noWait?: true;
+ json?: true;
+ status?: true;
+ result?: true;
+ resumeExecution?: true;
+ placement?: 'current_turn' | 'next_turn';
+ answers?: readonly (string | null)[];
prompt?: string;
stdinPrompt: boolean;
cwd?: string;
@@ -78,13 +92,15 @@ export interface MakaRunContext {
export interface MakaRunOutcome {
outcomeId: string;
- status: 'completed' | 'failed';
+ status: 'completed' | 'failed' | 'running';
finalOutput?: string;
failure?: { class: string; message?: string };
sandboxBoundary: 'none' | 'unresolved';
}
export interface MakaRunContextInput {
+ noWait?: boolean;
+ allowDetach?: boolean;
workspaceRoot: string;
cwd: string;
requestedConnectionSlug?: string;
@@ -99,6 +115,7 @@ export interface MakaRunContextInput {
}
export interface MakaRunDeps {
+ runSession?(options: MakaRunOptions, env: MakaRunEnvironmentDeps): Promise<MakaRunSessionResult>;
createContext(input: MakaRunContextInput): Promise<MakaRunContext>;
listSessions(workspaceRoot: string, hostProfileId?: string): Promise<SessionSummary[]>;
workspaceRoot(): string;
@@ -114,10 +131,12 @@ export interface MakaRunDeps {
newId(): string;
}
-export type MakaRunAdapter = Pick<MakaRunDeps, 'createContext' | 'listSessions'>;
+export type MakaRunAdapter = Pick<MakaRunDeps, 'createContext' | 'listSessions' | 'runSession'>;
export type MakaRunEnvironmentDeps = Omit<MakaRunDeps, keyof MakaRunAdapter>;
const VALUE_FLAGS = new Set([
+ 'session',
+ 'answer',
'cwd',
'connection',
'model',
@@ -130,13 +149,26 @@ const VALUE_FLAGS = new Set([
]);
const REPEATABLE_VALUE_FLAGS = new Set<string>();
-const BOOLEAN_FLAGS = new Set(['continue', 'yolo', 'graph']);
+const BOOLEAN_FLAGS = new Set([
+ 'continue',
+ 'yolo',
+ 'graph',
+ 'wait',
+ 'no-wait',
+ 'json',
+ 'status',
+ 'result',
+ 'queue',
+ 'steer',
+]);
export function parseMakaRunArgs(argv: readonly string[]): ParseMakaRunArgsResult {
const positional: string[] = [];
const flags = new Map<string, string>();
const booleanFlags = new Set<string>();
let literal = false;
+ const optionArgs = argv.includes('--') ? argv.slice(0, argv.indexOf('--')) : argv;
+ const selectsSession = optionArgs.includes('--session');
for (let index = 0; index < argv.length; index += 1) {
const arg = argv[index]!;
if (!literal && (arg === '--help' || arg === '-h')) return { kind: 'help' };
@@ -146,7 +178,7 @@ export function parseMakaRunArgs(argv: readonly string[]): ParseMakaRunArgsResul
}
if (!literal && arg.startsWith('--')) {
const name = arg.slice(2);
- if (BOOLEAN_FLAGS.has(name)) {
+ if (BOOLEAN_FLAGS.has(name) || (name === 'resume' && selectsSession)) {
if (booleanFlags.has(name)) return { kind: 'error', message: `option repeated: ${arg}` };
booleanFlags.add(name);
continue;
@@ -172,6 +204,64 @@ export function parseMakaRunArgs(argv: readonly string[]): ParseMakaRunArgsResul
if (positional.length > 1) {
return { kind: 'error', message: 'maka run accepts at most one positional prompt' };
}
+ const sessionId = flags.get('session');
+ const resumeExecution = booleanFlags.has('resume');
+ if (sessionId !== undefined && !sessionId.trim())
+ return { kind: 'error', message: '--session requires a non-empty Session ID' };
+ if (booleanFlags.has('wait') && booleanFlags.has('no-wait')) {
+ return { kind: 'error', message: '--wait and --no-wait cannot be used together' };
+ }
+ if (booleanFlags.has('queue') && booleanFlags.has('steer')) {
+ return { kind: 'error', message: '--queue and --steer cannot be used together' };
+ }
+ const actions =
+ Number(booleanFlags.has('status')) +
+ Number(booleanFlags.has('result')) +
+ Number(flags.has('answer')) +
+ Number(resumeExecution);
+ if (actions > 1)
+ return { kind: 'error', message: 'choose one of --status, --result, --answer or --resume' };
+ if (!sessionId && (actions || booleanFlags.has('queue') || booleanFlags.has('steer'))) {
+ return { kind: 'error', message: 'session operations require --session <session-id>' };
+ }
+ if (sessionId && (flags.has('resume') || booleanFlags.has('continue') || flags.has('project'))) {
+ return {
+ kind: 'error',
+ message: '--session cannot be combined with legacy --resume <id>, --continue or --project',
+ };
+ }
+ if (actions && positional.length)
+ return { kind: 'error', message: 'session actions do not accept a prompt' };
+ if ((booleanFlags.has('queue') || booleanFlags.has('steer')) && (!positional.length || actions)) {
+ return { kind: 'error', message: '--queue and --steer require a prompt' };
+ }
+ if (
+ (booleanFlags.has('status') || booleanFlags.has('result')) &&
+ (booleanFlags.has('wait') || booleanFlags.has('no-wait'))
+ ) {
+ return { kind: 'error', message: '--status and --result do not accept wait options' };
+ }
+ if (booleanFlags.has('graph') && booleanFlags.has('no-wait')) {
+ return { kind: 'error', message: '--graph must wait for its final result' };
+ }
+ let answers: (string | null)[] | undefined;
+ if (flags.has('answer')) {
+ try {
+ const value: unknown = JSON.parse(flags.get('answer')!);
+ if (
+ !Array.isArray(value) ||
+ value.length === 0 ||
+ value.some((v) => v !== null && typeof v !== 'string')
+ )
+ throw new Error();
+ answers = value;
+ } catch {
+ return {
+ kind: 'error',
+ message: '--answer expects a JSON array of strings or null, e.g. ["Yes"]',
+ };
+ }
+ }
const prompt = positional[0];
const timeout = flags.get('timeout');
const maxSteps = flags.get('max-steps');
@@ -205,6 +295,19 @@ export function parseMakaRunArgs(argv: readonly string[]): ParseMakaRunArgsResul
return {
kind: 'run',
options: {
+ ...(sessionId !== undefined ? { sessionId } : {}),
+ ...(booleanFlags.has('wait') ? { wait: true as const } : {}),
+ ...(booleanFlags.has('no-wait') ? { noWait: true as const } : {}),
+ ...(booleanFlags.has('json') ? { json: true as const } : {}),
+ ...(booleanFlags.has('status') ? { status: true as const } : {}),
+ ...(booleanFlags.has('result') ? { result: true as const } : {}),
+ ...(resumeExecution ? { resumeExecution: true as const } : {}),
+ ...(answers ? { answers } : {}),
+ ...(booleanFlags.has('steer')
+ ? { placement: 'current_turn' as const }
+ : booleanFlags.has('queue')
+ ? { placement: 'next_turn' as const }
+ : {}),
...(prompt !== undefined && prompt !== '-' ? { prompt } : {}),
stdinPrompt: prompt === '-',
...(flags.get('cwd') !== undefined ? { cwd: flags.get('cwd') } : {}),
@@ -239,6 +342,23 @@ export async function runMakaTextCliCore(
return 2;
}
+ if (parsed.options.sessionId) {
+ try {
+ if (!deps.runSession)
+ throw new Error('Session operations are unavailable in this CLI runtime');
+ const options = { ...parsed.options };
+ // Queries and answers must not block reading an unrelated stdin stream.
+ if (options.prompt !== undefined || options.stdinPrompt)
+ options.prompt = await resolveRunPrompt(options, deps);
+ const result = await deps.runSession(options, deps);
+ writeSessionResult(result, deps, options.json === true);
+ return result.status === 'failed' ? 1 : 0;
+ } catch (error) {
+ deps.writeStderr(`maka run: ${errorMessage(error)}\n`);
+ return error instanceof RunSessionInterrupted ? 130 : 1;
+ }
+ }
+
let prompt: string;
let selection: Awaited<ReturnType<typeof selectMakaRunSession>>;
const workspaceRoot = deps.workspaceRoot();
@@ -279,6 +399,7 @@ export async function runMakaTextCliCore(
let context: MakaRunContext;
try {
context = await deps.createContext({
+ ...(parsed.options.noWait ? { noWait: true } : {}),
workspaceRoot,
cwd: selection.cwd,
...(selection.kind === 'existing' || parsed.options.connection
@@ -351,6 +472,7 @@ export async function runMakaTextCliCore(
let interrupted = false;
let timedOut = false;
let streamFailed = false;
+ let question: RunQuestionAvailable | undefined;
let stopPromise: Promise<void> | undefined;
let resolveStopSignal: (() => void) | undefined;
const stopSignal = new Promise<void>((resolve) => {
@@ -416,10 +538,11 @@ export async function runMakaTextCliCore(
}
await stopPromise;
} catch (error) {
- streamFailed = true;
+ question = error instanceof RunQuestionAvailable ? error : undefined;
+ streamFailed = !question;
graphActivity?.release();
await stopPromise?.catch(() => undefined);
- if (!interrupted && !timedOut) {
+ if (!interrupted && !timedOut && !question) {
deps.writeStderr(`maka run: ${errorMessage(error)}\n`);
}
} finally {
@@ -433,21 +556,43 @@ export async function runMakaTextCliCore(
deps.writeStderr(`maka run: timed out after ${parsed.options.timeoutMs}ms\n`);
return 1;
}
+ if (question) {
+ writeSessionResult(question.result(), deps, true);
+ return 0;
+ }
if (streamFailed) return 1;
if (boundaryFailure) return 1;
if (!outcome) {
deps.writeStderr('maka run: runtime produced no outcome\n');
return 1;
}
+ if (outcome.status === 'running') {
+ writeSessionResult({ sessionId: session.id, status: 'running' }, deps, true);
+ return 0;
+ }
if (outcome.status !== 'completed' || outcome.finalOutput === undefined) {
const detail = outcome.failure?.message ?? outcome.failure?.class ?? 'runtime failure';
deps.writeStderr(`maka run: ${detail}\n`);
return 1;
}
- deps.writeStdout(withTrailingNewline(outcome.finalOutput));
+ writeSessionResult(
+ { sessionId: session.id, status: 'completed', finalOutput: outcome.finalOutput },
+ deps,
+ parsed.options.json === true,
+ );
return 0;
}
+export function writeSessionResult(
+ result: MakaRunSessionResult,
+ deps: Pick<MakaRunDeps, 'writeStdout' | 'writeStderr'>,
+ json: boolean,
+): void {
+ if (json || result.status !== 'completed') deps.writeStdout(`${JSON.stringify(result)}\n`);
+ else deps.writeStdout(withTrailingNewline(result.finalOutput ?? ''));
+ if (result.status === 'failed' && result.error) deps.writeStderr(`maka run: ${result.error}\n`);
+}
+
async function resolveRunPrompt(options: MakaRunOptions, deps: MakaRunDeps): Promise<string> {
const shouldReadStdin = options.stdinPrompt || !deps.stdinIsTTY();
const stdin = shouldReadStdin ? await deps.readStdin() : '';
@@ -484,7 +629,17 @@ function makaRunHelpText(cliCommand: string): string {
' --timeout <seconds> Invocation timeout',
' --max-steps <count> Tool-step cap',
' --yolo Give this task full access to your files and network',
- ' --resume <session-id> Continue an explicit compatible task',
+ ' --session <session-id> Select a Session (without a prompt: show its status)',
+ ' --wait Wait for this message/current Turn; return questions early',
+ ' --no-wait Submit and return (Host-owned execution only)',
+ ' --status Read current Session status without executing',
+ ' --result Read the latest Turn result without waiting',
+ ' --answer <json-array> Answer the current AskUser, e.g. ["Yes",null]',
+ ' --queue Queue this prompt as its own next Turn (default)',
+ ' --steer Add this prompt to the current execution',
+ ' --json Return a structured result, including the Session ID',
+ ' --session <id> --resume Safely resume stopped execution',
+ ' --resume <session-id> Legacy Session selector (prefer --session)',
' --continue Continue the latest compatible task for cwd',
' --graph Run this turn in Graph Mode and wait for graph completion',
' -h, --help Show help',
diff --git a/packages/cli/src/run-session-command.ts b/packages/cli/src/run-session-command.ts
new file mode 100644
index 0000000000..7eb02397f8
--- /dev/null
+++ b/packages/cli/src/run-session-command.ts
@@ -0,0 +1,391 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { StoredMessage } from '@maka/core/session';
+import type { RuntimeHostConnection } from '@maka/runtime-host/client';
+import { updateRuntimeHostSession } from './runtime-host-session-update.js';
+import type {
+ InteractionPendingSnapshot,
+ SessionContinuitySnapshot,
+} from '@maka/runtime-host/protocol';
+import { RuntimeHostSessionChannel } from './runtime-host-session-channel.js';
+import type { MakaRunEnvironmentDeps, MakaRunOptions } from './run-command-core.js';
+import { RunSessionInterrupted, type MakaRunSessionResult } from './run-session-result.js';
+import { TurnOutcomeClassifier, acceptStoredTurn } from './run-turn-outcome.js';
+
+/** Session commands observe Host-owned execution; they never publish client MCP. */
+export async function runMakaSessionCommand(
+ connection: RuntimeHostConnection,
+ options: MakaRunOptions,
+ env: Pick<MakaRunEnvironmentDeps, 'onSigint' | 'newId'>,
+ allowDetach = true,
+): Promise<MakaRunSessionResult> {
+ const sessionId = options.sessionId!;
+ const executing =
+ options.prompt !== undefined || options.answers !== undefined || options.resumeExecution;
+ if (executing && !allowDetach)
+ throw new Error(
+ 'Session control execution requires a persistent installation; use the blocking legacy run path for temporary npx invocations',
+ );
+ // Never silently discard options that the old turn.start path supports.
+ if (
+ options.graph ||
+ options.connection ||
+ options.model ||
+ options.thinking !== undefined ||
+ options.cwd ||
+ options.projectId ||
+ options.timeoutMs
+ ) {
+ throw new Error(
+ 'Session commands use the saved configuration; use the blocking legacy --resume <id> path for configuration, cwd, timeout or Graph options',
+ );
+ }
+ if (
+ options.maxSteps !== undefined &&
+ (options.prompt === undefined || options.placement !== undefined)
+ ) {
+ throw new Error('--max-steps requires an idle Session prompt without --queue/--steer');
+ }
+ const cancelled = new AbortController();
+ const removeSigint = env.onSigint(() => cancelled.abort(new RunSessionInterrupted()));
+ const observer = await SessionRunObserver.open(connection, sessionId, cancelled.signal).catch(
+ (error) => {
+ removeSigint();
+ throw error;
+ },
+ );
+ const assertActive = () => {
+ cancelled.signal.throwIfAborted();
+ observer.throwIfFailed();
+ };
+ try {
+ assertActive();
+ let target = observer.snapshot.rootTurn?.turnId;
+ let messageId: string | undefined;
+ if (executing) {
+ const boundary = await connection.request('session.execution_boundary.query', { sessionId });
+ assertActive();
+ if (boundary.kind === 'external')
+ throw new Error('This Session must be controlled by its owning harness');
+ if (options.yolo) {
+ await updateRuntimeHostSession(
+ connection,
+ sessionId,
+ (current) =>
+ connection.request('session.configuration.update', {
+ sessionId,
+ expectedRevision: current.revision,
+ patch: { permissionMode: 'bypass' },
+ }),
+ { operation: 'session.configuration.update', assertRequestAllowed: assertActive },
+ );
+ } else if (boundary.kind === 'bypass') {
+ throw new Error(`executing a full-access session ${sessionId} requires --yolo`);
+ }
+ }
+ assertActive();
+ if (options.answers !== undefined) {
+ const pending = observer.snapshot.interactions.pending.filter(
+ (p) => p.request.kind === 'question',
+ );
+ if (pending.length !== 1)
+ throw new Error(
+ `Expected one pending AskUser batch, found ${pending.length}; no answer was submitted`,
+ );
+ const question = pending[0]!;
+ target = question.turnId;
+ // Use the observed identity exactly once. A dispatched answer is never retried here.
+ await connection.request('interaction.answer', {
+ sessionId,
+ interactionId: question.interactionId,
+ answer: { kind: 'question', answers: options.answers },
+ });
+ observer.ignoreAnswered(question.interactionId);
+ } else if (options.resumeExecution) {
+ const plan = await connection.request('turn.resume.query', { sessionId });
+ assertActive();
+ if (plan.disposition !== 'ready') throw new Error(`Execution cannot resume: ${plan.reason}`);
+ const resumed = await connection.request('turn.resume.start', {
+ sessionId,
+ turnId: env.newId(),
+ sourceRunId: plan.sourceRunId,
+ sourceRuntimeEventHighWater: plan.sourceRuntimeEventHighWater,
+ });
+ if (resumed.kind !== 'started')
+ throw new Error(`Execution cannot resume: ${resumed.plan.reason}`);
+ target = resumed.turn.turnId;
+ } else if (options.prompt !== undefined) {
+ if (options.maxSteps !== undefined) {
+ const root = observer.snapshot.rootTurn;
+ if (root && !terminal(root.status))
+ throw new Error('--max-steps cannot be applied to a queued or active Turn');
+ const started = await connection.request('turn.start', {
+ sessionId,
+ turnId: env.newId(),
+ content: { text: options.prompt },
+ maxSteps: options.maxSteps,
+ });
+ if (started.kind !== 'started') throw new Error('Turn did not start');
+ target = started.turn.turnId;
+ } else {
+ messageId = env.newId();
+ const submitted = await connection.request('turn.message.submit', {
+ originHostEpoch: connection.hostEpoch,
+ sessionId,
+ messageId,
+ content: { text: options.prompt },
+ placement: options.placement ?? 'next_turn',
+ });
+ if (submitted.disposition === 'blocked')
+ throw new Error('Message was blocked before execution');
+ if (submitted.disposition === 'turn_started') target = submitted.turnId;
+ else if (submitted.disposition === 'steering') {
+ // Steering targets the Turn selected on attachment; it does not own a new Turn.
+ if (!target)
+ throw new Error(
+ 'Steering was admitted but its observed Turn is unavailable; query Session status before continuing',
+ );
+ messageId = undefined;
+ } else target = undefined;
+ }
+ }
+ assertActive();
+ if (options.status || (!executing && !options.wait && !options.result))
+ return observer.status();
+ if (options.noWait) return { sessionId, status: 'running' };
+ if (!target && !messageId) target = await observer.latestTurn();
+ if (!target && !messageId) return { sessionId, status: 'idle' };
+ for (;;) {
+ const version = observer.version;
+ cancelled.signal.throwIfAborted();
+ observer.throwIfFailed();
+ if (messageId && !target) {
+ const execution = await connection.request('turn.message.execution.query', {
+ sessionId,
+ messageIds: [messageId],
+ });
+ const resolution = execution.resolutions[0];
+ if (resolution?.state === 'owned') target = resolution.turnId;
+ else if (resolution?.state === 'cancelled' || resolution?.state === 'not_admitted') {
+ throw new Error(`Submitted message is ${resolution.state}; it has no execution result`);
+ }
+ }
+ if (target) {
+ const questions = observer.questions(target);
+ if (questions.length) return observer.questionResult(questions);
+ const turn = await connection.request('turn.query', { sessionId, turnId: target });
+ if (terminal(turn.status)) {
+ const result = await observer.result(target);
+ if (result) return result;
+ if (options.result)
+ throw new Error('The terminal result is still being published; query it again');
+ await observer.changed(version);
+ continue;
+ }
+ const unsupported = observer.snapshot.interactions.pending.find(
+ (p) => p.turnId === target && p.request.kind !== 'question',
+ );
+ if (unsupported)
+ throw new Error(
+ `Pending ${unsupported.request.kind} interaction requires its owning interactive Client; MCP form continuation is not supported by maka run`,
+ );
+ if (options.result) return { sessionId, status: 'running' };
+ }
+ await observer.changed(version);
+ }
+ } finally {
+ removeSigint();
+ await observer.close();
+ }
+}
+
+function terminal(status: string): boolean {
+ return status === 'completed' || status === 'failed' || status === 'cancelled';
+}
+
+class SessionRunObserver {
+ readonly #connection: RuntimeHostConnection;
+ readonly #sessionId: string;
+ readonly #signal: AbortSignal;
+ #channel!: RuntimeHostSessionChannel;
+ #version = 0;
+ #failure: unknown;
+ #waiters = new Set<() => void>();
+ #answered = new Set<string>();
+
+ private constructor(connection: RuntimeHostConnection, sessionId: string, signal: AbortSignal) {
+ this.#connection = connection;
+ this.#sessionId = sessionId;
+ this.#signal = signal;
+ }
+
+ static async open(
+ connection: RuntimeHostConnection,
+ sessionId: string,
+ signal: AbortSignal,
+ ): Promise<SessionRunObserver> {
+ const observer = new SessionRunObserver(connection, sessionId, signal);
+ const changed = () => observer.#notify();
+ const opened = await RuntimeHostSessionChannel.open({
+ connection,
+ sessionId,
+ signal,
+ now: Date.now,
+ onTurnStarted: (turn) => {
+ // Drain live events so a long-running observation cannot back up the channel.
+ void (async () => {
+ for await (const _event of turn.events) {
+ /* results are reconciled from durable transcript */
+ }
+ })().catch((error) => {
+ observer.#failure = error;
+ changed();
+ });
+ },
+ onRuntimeResourceChanged: changed,
+ onInteractionPending: changed,
+ onInteractionResolved: changed,
+ onTranscriptSettlement: changed,
+ onTranscriptReplaced: changed,
+ onGoalChanged: changed,
+ onSnapshotChanged: changed,
+ onRecovered: changed,
+ onFailed: (error) => {
+ observer.#failure = error;
+ changed();
+ },
+ });
+ observer.#channel = opened.channel;
+ opened.channel.activate();
+ return observer;
+ }
+
+ get snapshot(): SessionContinuitySnapshot {
+ return this.#channel.snapshot;
+ }
+ get version(): number {
+ return this.#version;
+ }
+ throwIfFailed(): void {
+ if (this.#failure) throw this.#failure;
+ }
+ ignoreAnswered(id: string): void {
+ this.#answered.add(id);
+ }
+
+ questions(turnId?: string): InteractionPendingSnapshot[] {
+ return this.snapshot.interactions.pending.filter(
+ (p) =>
+ p.request.kind === 'question' &&
+ !this.#answered.has(p.interactionId) &&
+ (turnId === undefined || p.turnId === turnId),
+ );
+ }
+
+ questionResult(pending: readonly InteractionPendingSnapshot[]): MakaRunSessionResult {
+ return {
+ sessionId: this.#sessionId,
+ status: 'needs_input',
+ questions: pending.flatMap((p) => (p.request.kind === 'question' ? p.request.questions : [])),
+ ...(pending.length > 1 ? { pendingQuestionCount: pending.length } : {}),
+ };
+ }
+
+ status(): MakaRunSessionResult {
+ const pending = this.questions();
+ if (pending.length)
+ return {
+ ...this.questionResult(pending),
+ queued: this.snapshot.queue.followup.length + this.snapshot.queue.steering.length,
+ };
+ const root = this.snapshot.rootTurn;
+ return {
+ sessionId: this.#sessionId,
+ status: !root
+ ? 'idle'
+ : root.status === 'completed'
+ ? 'completed'
+ : terminal(root.status)
+ ? 'failed'
+ : 'running',
+ queued: this.snapshot.queue.followup.length + this.snapshot.queue.steering.length,
+ };
+ }
+
+ async latestTurn(): Promise<string | undefined> {
+ const result = await this.#connection.request('session.turn_landmarks.query', {
+ sessionId: this.#sessionId,
+ maxLandmarks: 1,
+ turnId: null,
+ });
+ return result.landmarks.at(-1)?.turnId;
+ }
+
+ async result(turnId: string): Promise<MakaRunSessionResult | undefined> {
+ const classifier = new TurnOutcomeClassifier(turnId);
+ await this.#channel.replayTranscript(async (messages: readonly StoredMessage[]) => {
+ acceptStoredTurn(classifier, messages, turnId);
+ }, this.#signal);
+ const outcome = classifier.outcome('pending');
+ if (!outcome) return undefined;
+ if (outcome.status === 'completed' && outcome.sandboxBoundary === 'none') {
+ return {
+ sessionId: this.#sessionId,
+ status: 'completed',
+ finalOutput: outcome.finalOutput ?? '',
+ };
+ }
+ return {
+ sessionId: this.#sessionId,
+ status: 'failed',
+ error:
+ outcome.sandboxBoundary === 'unresolved'
+ ? 'Turn encountered an unresolved sandbox boundary'
+ : (outcome.failure?.message ?? outcome.failure?.class ?? 'Turn failed'),
+ };
+ }
+
+ changed(version: number): Promise<void> {
+ this.#signal.throwIfAborted();
+ if (this.#version !== version) return Promise.resolve();
+ return new Promise((resolve, reject) => {
+ const done = () => {
+ this.#signal.removeEventListener('abort', abort);
+ this.#waiters.delete(done);
+ resolve();
+ };
+ const abort = () => {
+ this.#waiters.delete(done);
+ reject(this.#signal.reason);
+ };
+ this.#waiters.add(done);
+ this.#signal.addEventListener('abort', abort, { once: true });
+ });
+ }
+
+ #notify(): void {
+ this.#version++;
+ for (const wake of [...this.#waiters]) wake();
+ }
+
+ async close(): Promise<void> {
+ await this.#channel.close();
+ }
+}
diff --git a/packages/cli/src/run-session-result.ts b/packages/cli/src/run-session-result.ts
new file mode 100644
index 0000000000..91a05ac523
--- /dev/null
+++ b/packages/cli/src/run-session-result.ts
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { InteractionQuestion } from '@maka/core/interaction';
+import type { InteractionPendingSnapshot } from '@maka/runtime-host/protocol';
+
+export interface MakaRunSessionResult {
+ sessionId: string;
+ status: 'running' | 'needs_input' | 'completed' | 'failed' | 'idle';
+ finalOutput?: string;
+ error?: string;
+ questions?: readonly InteractionQuestion[];
+ pendingQuestionCount?: number;
+ queued?: number;
+}
+
+/** A normal handoff to the caller, not a failed or cancelled Host Turn. */
+export class RunQuestionAvailable extends Error {
+ constructor(readonly pending: InteractionPendingSnapshot) {
+ super('The Turn is waiting for an answer');
+ }
+
+ result(): MakaRunSessionResult {
+ return {
+ sessionId: this.pending.sessionId,
+ status: 'needs_input',
+ questions: this.pending.request.kind === 'question' ? this.pending.request.questions : [],
+ };
+ }
+}
+
+export class RunSessionInterrupted extends Error {
+ constructor() {
+ super('Session observation interrupted; Host execution was not stopped');
+ }
+}
diff --git a/packages/cli/src/run-turn-outcome.ts b/packages/cli/src/run-turn-outcome.ts
new file mode 100644
index 0000000000..379d38a20c
--- /dev/null
+++ b/packages/cli/src/run-turn-outcome.ts
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { failureClassFromCompleteStopReason, type SessionEvent } from '@maka/core/events';
+import type { StoredMessage } from '@maka/core/session';
+import type { MakaRunOutcome } from './run-command-core.js';
+
+type TurnOutcomeObservation =
+ | { readonly kind: 'output'; readonly text: string; readonly source: 'live' | 'stored' }
+ | {
+ readonly kind: 'terminal';
+ readonly update: 'replace' | 'if_unset';
+ readonly status: 'completed';
+ }
+ | {
+ readonly kind: 'terminal';
+ readonly update: 'replace' | 'if_unset';
+ readonly status: 'failed';
+ readonly failure: NonNullable<MakaRunOutcome['failure']>;
+ }
+ | {
+ readonly kind: 'tool_result';
+ readonly toolUseId: string;
+ readonly outcome: 'sandbox_failure' | 'success';
+ };
+
+type TerminalOutcomeObservation = Extract<TurnOutcomeObservation, { kind: 'terminal' }>;
+
+export class TurnOutcomeClassifier {
+ readonly #outcomeId: string;
+ readonly #unresolvedSandboxFailures = new Set<string>();
+ #finalOutput: string | undefined;
+ #finalOutputFromLive = false;
+ #terminal: TerminalOutcomeObservation | undefined;
+
+ constructor(outcomeId: string) {
+ this.#outcomeId = outcomeId;
+ }
+
+ accept(observation: TurnOutcomeObservation | undefined): void {
+ switch (observation?.kind) {
+ case undefined:
+ return;
+ case 'output':
+ // A subscriber is delivered in Host order, so the stream's last answer
+ // is the Turn's last answer. A transcript read stops wherever the Host
+ // had committed, so it can supply an answer the stream never carried
+ // but can never overrule one it did.
+ if (observation.source === 'stored' && this.#finalOutputFromLive) return;
+ this.#finalOutput = observation.text;
+ this.#finalOutputFromLive = observation.source === 'live';
+ return;
+ case 'terminal':
+ if (observation.update === 'replace' || this.#terminal === undefined) {
+ this.#terminal = observation;
+ }
+ return;
+ case 'tool_result': {
+ if (observation.outcome === 'sandbox_failure') {
+ this.#unresolvedSandboxFailures.add(observation.toolUseId);
+ }
+ // No clearing path: `maka run` denies every widening request, so the
+ // boundary cannot move mid-Turn and a later success cannot prove that
+ // a blocked call recovered. The failure stays unresolved to the end.
+ return;
+ }
+ }
+ }
+
+ outcome(incomplete: 'fail'): MakaRunOutcome;
+ outcome(incomplete: 'pending'): MakaRunOutcome | undefined;
+ outcome(incomplete: 'fail' | 'pending'): MakaRunOutcome | undefined {
+ const terminal = this.#terminal;
+ if (!terminal && incomplete === 'pending') return undefined;
+ const completed = terminal?.status === 'completed';
+ const sandboxBoundary = this.#unresolvedSandboxFailures.size > 0 ? 'unresolved' : 'none';
+ const failure =
+ terminal?.status === 'failed'
+ ? terminal.failure
+ : {
+ class: 'missing_terminal_event',
+ message: 'Turn ended unexpectedly',
+ };
+ return {
+ outcomeId: this.#outcomeId,
+ status: completed ? 'completed' : 'failed',
+ ...(completed && this.#finalOutput !== undefined ? { finalOutput: this.#finalOutput } : {}),
+ ...(!completed ? { failure } : {}),
+ sandboxBoundary,
+ };
+ }
+}
+
+export function observationFromSessionEvent(
+ event: SessionEvent,
+): TurnOutcomeObservation | undefined {
+ if (event.type === 'text_complete' && event.text.trim().length > 0) {
+ return { kind: 'output', text: event.text, source: 'live' };
+ }
+ if (event.type === 'error') {
+ return {
+ kind: 'terminal',
+ update: 'replace',
+ status: 'failed',
+ failure: { class: event.reason ?? event.code ?? 'runtime_error', message: event.message },
+ };
+ }
+ if (event.type === 'abort') {
+ return {
+ kind: 'terminal',
+ update: 'replace',
+ status: 'failed',
+ failure: { class: 'aborted', message: 'Turn was cancelled' },
+ };
+ }
+ if (event.type === 'complete') {
+ return observationFromCompleteEvent(event);
+ }
+ return event.type === 'tool_result' ? observationFromToolResult(event) : undefined;
+}
+
+function observationFromStoredMessage(message: StoredMessage): TurnOutcomeObservation | undefined {
+ if (message.type === 'assistant' && message.text.trim().length > 0) {
+ return { kind: 'output', text: message.text, source: 'stored' };
+ }
+ if (message.type === 'turn_state' && message.status === 'completed') {
+ return { kind: 'terminal', update: 'replace', status: 'completed' };
+ }
+ if (message.type === 'turn_state' && message.status === 'aborted') {
+ return {
+ kind: 'terminal',
+ update: 'replace',
+ status: 'failed',
+ failure: { class: 'aborted', message: 'Turn was cancelled' },
+ };
+ }
+ if (message.type === 'turn_state' && message.status === 'failed') {
+ return {
+ kind: 'terminal',
+ update: 'replace',
+ status: 'failed',
+ failure: {
+ class: message.errorClass ?? 'runtime_error',
+ message: message.failureMessage ?? 'Turn failed',
+ },
+ };
+ }
+ return message.type === 'tool_result' ? observationFromToolResult(message) : undefined;
+}
+
+function observationFromCompleteEvent(
+ event: Extract<SessionEvent, { type: 'complete' }>,
+): TerminalOutcomeObservation {
+ if (event.stopReason === 'user_stop') {
+ return {
+ kind: 'terminal',
+ update: 'if_unset',
+ status: 'failed',
+ failure: { class: 'aborted', message: 'Turn was cancelled' },
+ };
+ }
+ const failureClass = failureClassFromCompleteStopReason(event.stopReason);
+ return failureClass
+ ? {
+ kind: 'terminal',
+ update: 'if_unset',
+ status: 'failed',
+ failure: { class: failureClass },
+ }
+ : { kind: 'terminal', update: 'if_unset', status: 'completed' };
+}
+
+function observationFromToolResult(
+ result: Pick<Extract<SessionEvent, { type: 'tool_result' }>, 'content' | 'isError' | 'toolUseId'>,
+): TurnOutcomeObservation | undefined {
+ if (result.isError && result.content.kind === 'text' && result.content.sandboxFailure) {
+ return {
+ kind: 'tool_result',
+ toolUseId: result.toolUseId,
+ outcome: 'sandbox_failure',
+ };
+ }
+ return result.isError
+ ? undefined
+ : { kind: 'tool_result', toolUseId: result.toolUseId, outcome: 'success' };
+}
+
+export function outcomeFromStoredTurn(
+ messages: readonly StoredMessage[],
+ turnId: string,
+): MakaRunOutcome | undefined {
+ return classifierFromStoredTurn(messages, turnId, turnId).outcome('pending');
+}
+
+export function classifierFromStoredTurn(
+ messages: readonly StoredMessage[],
+ turnId: string,
+ outcomeId: string,
+): TurnOutcomeClassifier {
+ const classifier = new TurnOutcomeClassifier(outcomeId);
+ acceptStoredTurn(classifier, messages, turnId);
+ return classifier;
+}
+
+export function acceptStoredTurn(
+ classifier: TurnOutcomeClassifier,
+ messages: readonly StoredMessage[],
+ turnId: string,
+): void {
+ for (const message of messages) {
+ if (message.turnId === turnId) classifier.accept(observationFromStoredMessage(message));
+ }
+}
diff --git a/packages/cli/src/runtime-host-run-command.ts b/packages/cli/src/runtime-host-run-command.ts
index 9e5437d93f..5056df091f 100644
--- a/packages/cli/src/runtime-host-run-command.ts
+++ b/packages/cli/src/runtime-host-run-command.ts
@@ -17,7 +17,7 @@
* under the License.
*/
-import { failureClassFromCompleteStopReason, type SessionEvent } from '@maka/core/events';
+import { type SessionEvent } from '@maka/core/events';
import { findProjectByIdentity } from '@maka/core/project';
import { type StoredMessage } from '@maka/core/session';
import type { UserMessageInput } from '@maka/core/runtime-inputs';
@@ -49,6 +49,7 @@ import {
} from './run-command-core.js';
import {
connectRuntimeHostCli,
+ connectRuntimeHostCliConnection,
resolveRuntimeHostCliTarget,
type RuntimeHostCliConnectionContext,
} from './runtime-host-cli-context.js';
@@ -65,6 +66,17 @@ import {
} from './runtime-host-task-readiness.js';
import { resolveMakaClientDataRoot } from './workspace-root.js';
+import {
+ TurnOutcomeClassifier,
+ observationFromSessionEvent,
+ outcomeFromStoredTurn,
+ classifierFromStoredTurn,
+ acceptStoredTurn,
+} from './run-turn-outcome.js';
+import { RunQuestionAvailable } from './run-session-result.js';
+import { isTemporaryNpxInstallation } from './runtime-host-cli-installation.js';
+import { runMakaSessionCommand } from './run-session-command.js';
+
const GRAPH_POLL_INTERVAL_MS = 25;
export interface RuntimeHostRunCommandDeps {
@@ -115,6 +127,23 @@ export async function runRuntimeHostTextCli(
return await commandDeps.run(
argv,
{
+ runSession: async (sessionOptions, env) => {
+ const context = await connectRuntimeHostCliConnection({
+ rootPath: env.workspaceRoot(),
+ ...(sessionOptions.hostProfileId ? { profileId: sessionOptions.hostProfileId } : {}),
+ clientDataRoot: options.clientDataRoot,
+ });
+ try {
+ return await runMakaSessionCommand(
+ context.connection,
+ sessionOptions,
+ env,
+ context.profile.kind !== 'local' || !(await isTemporaryNpxInstallation()),
+ );
+ } finally {
+ await context.close();
+ }
+ },
listSessions: async (rootPath, hostProfileId) =>
runtimeHostSessionSummaries(
await readRuntimeHostSessions((await connect(rootPath, hostProfileId)).connection),
@@ -197,6 +226,8 @@ export function createRuntimeHostRunContext(
input.enableAgentGraph === true,
input.sessionCwdOverride,
input.maxSteps,
+ input.noWait === true,
+ input.allowDetach !== false,
);
return {
runtime,
@@ -249,7 +280,11 @@ async function prepareRuntimeHostRunInput(
`Task is not ready:\n${formatRuntimeHostCliTaskBlockers(snapshot, cliCommand)}`,
);
}
- return preparedInput;
+ return {
+ ...preparedInput,
+ allowDetach:
+ runtimeHostProfileUsesHostWorkspace(profile.kind) || !(await isTemporaryNpxInstallation()),
+ };
}
type ActiveRuntimeHostTurn = {
@@ -267,6 +302,8 @@ class RuntimeHostRunRuntime implements MakaRunRuntime {
readonly #graphEnabled: boolean;
readonly #sessionCwdOverride: MakaRunContextInput['sessionCwdOverride'];
readonly #maxSteps: number | undefined;
+ readonly #noWait: boolean;
+ readonly #allowDetach: boolean;
readonly #unsubscribeTranscriptReplacements: () => void;
#sessionId: string | undefined;
#activeTurn: ActiveRuntimeHostTurn | undefined;
@@ -292,6 +329,8 @@ class RuntimeHostRunRuntime implements MakaRunRuntime {
graphEnabled: boolean,
sessionCwdOverride: MakaRunContextInput['sessionCwdOverride'],
maxSteps: number | undefined,
+ noWait = false,
+ allowDetach = true,
) {
this.#connection = connection;
this.#driver = driver;
@@ -300,8 +339,12 @@ class RuntimeHostRunRuntime implements MakaRunRuntime {
this.#graphEnabled = graphEnabled;
this.#sessionCwdOverride = sessionCwdOverride;
this.#maxSteps = maxSteps;
- this.#interactions = new NonInteractiveInteractionController(driver, (pending) =>
- this.#stopForInteraction(pending),
+ this.#noWait = noWait;
+ this.#allowDetach = allowDetach;
+ this.#interactions = new NonInteractiveInteractionController(
+ driver,
+ (pending) => this.#stopForInteraction(pending),
+ () => this.#allowDetach && !this.#graphEnabled && !this.#mcp?.ownsServers,
);
this.#unsubscribeTranscriptReplacements = driver.subscribeTranscriptReplacements(
(sessionId, turnId, messages) => {
@@ -326,6 +369,14 @@ class RuntimeHostRunRuntime implements MakaRunRuntime {
await this.#attach(sessionId);
await this.#mcp?.ready();
if (this.#stopRequested) throw new Error('Turn was cancelled before start');
+ if (this.#noWait && !this.#allowDetach)
+ throw new Error(
+ '--no-wait requires a persistent installation; temporary npx invocations may own the Host lifetime',
+ );
+ if (this.#noWait && this.#mcp?.ownsServers)
+ throw new Error(
+ '--no-wait cannot detach client-owned MCP servers; use the blocking run path',
+ );
if (input.turnOrchestration?.mode === 'graph') {
this.#graphAdmissionTurnIds = graphSupervisorTurnIds(await this.#driver.readMessages());
}
@@ -488,6 +539,14 @@ class RuntimeHostRunRuntime implements MakaRunRuntime {
turn: MakaPreparedSessionTurn,
active: ActiveRuntimeHostTurn,
): AsyncIterable<SessionEvent> {
+ if (this.#noWait) {
+ await this.#observer?.({
+ outcomeId: active.runId,
+ status: 'running',
+ sandboxBoundary: 'none',
+ });
+ return;
+ }
const events = turn.events[Symbol.asyncIterator]();
for (;;) {
const next = await this.#interactions.race(events.next());
@@ -604,184 +663,6 @@ function runtimeHostSessionSummaries(items: readonly SessionCatalogItem[]): Sess
return items.flatMap((item) => ('kind' in item ? [] : [projectSessionCatalogSummary(item)]));
}
-type TurnOutcomeObservation =
- | { readonly kind: 'output'; readonly text: string; readonly source: 'live' | 'stored' }
- | {
- readonly kind: 'terminal';
- readonly update: 'replace' | 'if_unset';
- readonly status: 'completed';
- }
- | {
- readonly kind: 'terminal';
- readonly update: 'replace' | 'if_unset';
- readonly status: 'failed';
- readonly failure: NonNullable<MakaRunOutcome['failure']>;
- }
- | {
- readonly kind: 'tool_result';
- readonly toolUseId: string;
- readonly outcome: 'sandbox_failure' | 'success';
- };
-
-type TerminalOutcomeObservation = Extract<TurnOutcomeObservation, { kind: 'terminal' }>;
-
-class TurnOutcomeClassifier {
- readonly #outcomeId: string;
- readonly #unresolvedSandboxFailures = new Set<string>();
- #finalOutput: string | undefined;
- #finalOutputFromLive = false;
- #terminal: TerminalOutcomeObservation | undefined;
-
- constructor(outcomeId: string) {
- this.#outcomeId = outcomeId;
- }
-
- accept(observation: TurnOutcomeObservation | undefined): void {
- switch (observation?.kind) {
- case undefined:
- return;
- case 'output':
- // A subscriber is delivered in Host order, so the stream's last answer
- // is the Turn's last answer. A transcript read stops wherever the Host
- // had committed, so it can supply an answer the stream never carried
- // but can never overrule one it did.
- if (observation.source === 'stored' && this.#finalOutputFromLive) return;
- this.#finalOutput = observation.text;
- this.#finalOutputFromLive = observation.source === 'live';
- return;
- case 'terminal':
- if (observation.update === 'replace' || this.#terminal === undefined) {
- this.#terminal = observation;
- }
- return;
- case 'tool_result': {
- if (observation.outcome === 'sandbox_failure') {
- this.#unresolvedSandboxFailures.add(observation.toolUseId);
- }
- // No clearing path: `maka run` denies every widening request, so the
- // boundary cannot move mid-Turn and a later success cannot prove that
- // a blocked call recovered. The failure stays unresolved to the end.
- return;
- }
- }
- }
-
- outcome(incomplete: 'fail'): MakaRunOutcome;
- outcome(incomplete: 'pending'): MakaRunOutcome | undefined;
- outcome(incomplete: 'fail' | 'pending'): MakaRunOutcome | undefined {
- const terminal = this.#terminal;
- if (!terminal && incomplete === 'pending') return undefined;
- const completed = terminal?.status === 'completed';
- const sandboxBoundary = this.#unresolvedSandboxFailures.size > 0 ? 'unresolved' : 'none';
- const failure =
- terminal?.status === 'failed'
- ? terminal.failure
- : {
- class: 'missing_terminal_event',
- message: 'Turn ended unexpectedly',
- };
- return {
- outcomeId: this.#outcomeId,
- status: completed ? 'completed' : 'failed',
- ...(completed && this.#finalOutput !== undefined ? { finalOutput: this.#finalOutput } : {}),
- ...(!completed ? { failure } : {}),
- sandboxBoundary,
- };
- }
-}
-
-function observationFromSessionEvent(event: SessionEvent): TurnOutcomeObservation | undefined {
- if (event.type === 'text_complete' && event.text.trim().length > 0) {
- return { kind: 'output', text: event.text, source: 'live' };
- }
- if (event.type === 'error') {
- return {
- kind: 'terminal',
- update: 'replace',
- status: 'failed',
- failure: { class: event.reason ?? event.code ?? 'runtime_error', message: event.message },
- };
- }
- if (event.type === 'abort') {
- return {
- kind: 'terminal',
- update: 'replace',
- status: 'failed',
- failure: { class: 'aborted', message: 'Turn was cancelled' },
- };
- }
- if (event.type === 'complete') {
- return observationFromCompleteEvent(event);
- }
- return event.type === 'tool_result' ? observationFromToolResult(event) : undefined;
-}
-
-function observationFromStoredMessage(message: StoredMessage): TurnOutcomeObservation | undefined {
- if (message.type === 'assistant' && message.text.trim().length > 0) {
- return { kind: 'output', text: message.text, source: 'stored' };
- }
- if (message.type === 'turn_state' && message.status === 'completed') {
- return { kind: 'terminal', update: 'replace', status: 'completed' };
- }
- if (message.type === 'turn_state' && message.status === 'aborted') {
- return {
- kind: 'terminal',
- update: 'replace',
- status: 'failed',
- failure: { class: 'aborted', message: 'Turn was cancelled' },
- };
- }
- if (message.type === 'turn_state' && message.status === 'failed') {
- return {
- kind: 'terminal',
- update: 'replace',
- status: 'failed',
- failure: {
- class: message.errorClass ?? 'runtime_error',
- message: 'Agent Graph final Turn failed',
- },
- };
- }
- return message.type === 'tool_result' ? observationFromToolResult(message) : undefined;
-}
-
-function observationFromCompleteEvent(
- event: Extract<SessionEvent, { type: 'complete' }>,
-): TerminalOutcomeObservation {
- if (event.stopReason === 'user_stop') {
- return {
- kind: 'terminal',
- update: 'if_unset',
- status: 'failed',
- failure: { class: 'aborted', message: 'Turn was cancelled' },
- };
- }
- const failureClass = failureClassFromCompleteStopReason(event.stopReason);
- return failureClass
- ? {
- kind: 'terminal',
- update: 'if_unset',
- status: 'failed',
- failure: { class: failureClass },
- }
- : { kind: 'terminal', update: 'if_unset', status: 'completed' };
-}
-
-function observationFromToolResult(
- result: Pick<Extract<SessionEvent, { type: 'tool_result' }>, 'content' | 'isError' | 'toolUseId'>,
-): TurnOutcomeObservation | undefined {
- if (result.isError && result.content.kind === 'text' && result.content.sandboxFailure) {
- return {
- kind: 'tool_result',
- toolUseId: result.toolUseId,
- outcome: 'sandbox_failure',
- };
- }
- return result.isError
- ? undefined
- : { kind: 'tool_result', toolUseId: result.toolUseId, outcome: 'success' };
-}
-
function graphSupervisorTurnIds(messages: readonly StoredMessage[]): Set<string> {
return new Set(
messages.flatMap((message) =>
@@ -804,36 +685,10 @@ function lastNewGraphSupervisorTurnId(
)?.turnId;
}
-function outcomeFromStoredTurn(
- messages: readonly StoredMessage[],
- turnId: string,
-): MakaRunOutcome | undefined {
- return classifierFromStoredTurn(messages, turnId, turnId).outcome('pending');
-}
-
-function classifierFromStoredTurn(
- messages: readonly StoredMessage[],
- turnId: string,
- outcomeId: string,
-): TurnOutcomeClassifier {
- const classifier = new TurnOutcomeClassifier(outcomeId);
- acceptStoredTurn(classifier, messages, turnId);
- return classifier;
-}
-
-function acceptStoredTurn(
- classifier: TurnOutcomeClassifier,
- messages: readonly StoredMessage[],
- turnId: string,
-): void {
- for (const message of messages) {
- if (message.turnId === turnId) classifier.accept(observationFromStoredMessage(message));
- }
-}
-
class NonInteractiveInteractionController {
readonly #driver: RuntimeHostMakaSessionDriver;
readonly #stop: (pending: InteractionPendingSnapshot) => Promise<void>;
+ readonly #canReturnQuestion: () => boolean;
readonly #handled = new Set<string>();
readonly #tasks = new Set<Promise<void>>();
readonly #unsubscribe: () => void;
@@ -843,9 +698,11 @@ class NonInteractiveInteractionController {
constructor(
driver: RuntimeHostMakaSessionDriver,
stop: (pending: InteractionPendingSnapshot) => Promise<void>,
+ canReturnQuestion: () => boolean,
) {
this.#driver = driver;
this.#stop = stop;
+ this.#canReturnQuestion = canReturnQuestion;
this.#unsubscribe = driver.subscribePendingInteractions((pending) => this.#accept(pending));
}
@@ -898,6 +755,9 @@ class NonInteractiveInteractionController {
});
return;
}
+ if (pending.request.kind === 'question' && this.#canReturnQuestion()) {
+ throw new RunQuestionAvailable(pending);
+ }
await this.#stop(pending);
throw new Error(
pending.request.kind === 'question'
diff --git a/packages/cli/src/runtime-host-run-mcp.ts b/packages/cli/src/runtime-host-run-mcp.ts
index 29a70b2d52..0c2ab6ba46 100644
--- a/packages/cli/src/runtime-host-run-mcp.ts
+++ b/packages/cli/src/runtime-host-run-mcp.ts
@@ -152,6 +152,10 @@ export class RuntimeHostRunMcp {
}
}
+ get ownsServers(): boolean {
+ return this.#enabledServerIds.length > 0;
+ }
+
close(): Promise<void> {
this.#closeTask ??= this.#close(true);
return this.#closeTask;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org