From 363f39a39a01ee15ea60789db4bfd15c8ac2406c Mon Sep 17 00:00:00 2001 From: Tarik02 Date: Mon, 24 Aug 2026 01:32:01 +0300 Subject: [PATCH 01/11] feat: add one-shot ask command --- .changeset/quiet-answers-land.md | 5 + README.md | 18 ++ skills/t3code-cli/SKILL.md | 13 + skills/t3code-cli/reference/commands.md | 32 ++ src/application/service.ts | 3 + src/application/threads.ts | 22 +- src/cli/app.ts | 2 + src/cli/ask-lifecycle.ts | 285 +++++++++++++++++ src/cli/ask.ts | 407 ++++++++++++++++++++++++ src/cli/error.ts | 54 ++++ 10 files changed, 840 insertions(+), 1 deletion(-) create mode 100644 .changeset/quiet-answers-land.md create mode 100644 src/cli/ask-lifecycle.ts create mode 100644 src/cli/ask.ts diff --git a/.changeset/quiet-answers-land.md b/.changeset/quiet-answers-land.md new file mode 100644 index 0000000..878b6fb --- /dev/null +++ b/.changeset/quiet-answers-land.md @@ -0,0 +1,5 @@ +--- +"t3code-cli": minor +--- + +add a one-shot ask command for projects and existing threads diff --git a/README.md b/README.md index 195f16a..3bf6197 100644 --- a/README.md +++ b/README.md @@ -177,6 +177,23 @@ Lists available provider models. Use `--all` to include hidden or unavailable en ## Thread Workflow +### Asking One Question + +`ask` waits for one answer and prints only the assistant text by default: + +```sh +t3cli ask "Which package owns this API?" --project /path/to/project +t3cli ask "Summarize the decision" --thread +``` + +Without `--thread`, it creates a thread and archives it after a successful answer. With +`--thread`, it leaves the existing thread active unless `--archive` is set explicitly. +`--archive` accepts `never`, `always`, `on-success`, or `on-failure`. + +Existing busy threads fail by default. Use `--busy queue` to wait or `--busy steer` to send the +question immediately. `--timeout 5m` limits the queue and response wait. Unlike other +thread-scoped commands, `ask` uses only an explicit `--thread` and ignores `T3CODE_THREAD_ID`. + ### Starting Threads ```sh @@ -304,6 +321,7 @@ Most commands support: Thread commands also support `ndjson` for streaming: ```sh +t3cli ask "task" --format ndjson t3cli start "task" --format ndjson --wait t3cli wait --format ndjson ``` diff --git a/skills/t3code-cli/SKILL.md b/skills/t3code-cli/SKILL.md index 7758f71..f977a6b 100644 --- a/skills/t3code-cli/SKILL.md +++ b/skills/t3code-cli/SKILL.md @@ -74,6 +74,19 @@ t3cli start "task" --format json --wait t3cli start "task" --format ndjson --wait # stream events ``` +**Ask once** + +Use `ask` when only the final answer is needed. It creates and archives a temporary thread by +default, or sends to an explicit existing thread without archiving it: + +```sh +t3cli ask "question" --project +t3cli ask "follow-up" --thread +``` + +`ask` ignores `T3CODE_THREAD_ID`; pass `--thread` intentionally. Plain output is answer text only. +Use `--format json` for a result object or `--format ndjson` for streamed events. + **Follow-up** ```sh diff --git a/skills/t3code-cli/reference/commands.md b/skills/t3code-cli/reference/commands.md index 6567220..3404ec8 100644 --- a/skills/t3code-cli/reference/commands.md +++ b/skills/t3code-cli/reference/commands.md @@ -98,6 +98,14 @@ Mutations dispatch `project.meta.update` with the next full scripts array and wa t3cli list [--project ] [--archived | --all] [--format json] t3cli search [--limit <1-50>] [--format auto|human|json] +t3cli ask [message] + [--project ] [--thread ] [--force|-f] [--stdin] + [--title ] [--worktree <path>] [--provider <name>] [--model <id>] + [--option key=value] [--reasoning-effort <v>] [--effort <v>] [--fast-mode] [--thinking] + [--archive never|always|on-success|on-failure] + [--busy fail|queue|steer] [--timeout <duration>] + [--format auto|human|json|ndjson] + t3cli start [message] [--project <ref>] [--stdin] [--title <title>] [--worktree <path>] [--provider <name>] [--model <id>] @@ -115,6 +123,25 @@ t3cli transcript [--thread <id>] [--limit N] t3cli wait [--thread <id>] [--format auto|human|ndjson] ``` +`ask` always waits for the turn it starts. Without `--thread`, it creates a thread and defaults to +`--archive on-success`. With an explicit `--thread`, it defaults to `--archive never`. An explicit +archive policy applies to either target. Archive failures produce a warning and remain visible in +structured output without changing a successful exit status. + +`--busy fail` is the default for an active existing thread. `queue` waits for the active turn, +while `steer` dispatches immediately. Busy checks are best effort because the server does not offer +an atomic busy-policy condition. Existing archived threads and threads with pending approval or +user-input requests are rejected. + +`--timeout` accepts positive durations such as `30s`, `5m`, and `1h`; omitting it waits without a +limit. A timeout or local interruption stops a turn started by `ask` when ownership can be +confirmed, then applies the failure archive policy. Time spent waiting under `--busy queue` counts +toward the timeout, but a timeout before dispatch does not mutate the existing turn. + +`--title`, `--worktree`, `--provider`, and `--model` apply only when creating a thread and are +rejected with `--thread`. Model option flags apply to both modes. `ask` never reads +`T3CODE_THREAD_ID`; selecting an existing thread requires `--thread`. + `transcript` loads the latest 10 user turns by default. Older-page requests default to 20 turns. JSON output includes `page.beforeCursor` and `page.hasMore`; pass the cursor to `--before-cursor` to load the next older page. `--turn-limit` sets either page size. `--all` loads the full thread and @@ -185,6 +212,7 @@ t3cli thread callback --from <thread-id> --prompt <message> [--thread <id>] [--b | Commands | `--format` | Agent default | | --------------- | ----------------------------- | ------------------------------- | | Most | `auto` \| `human` \| `json` | `json` | +| `ask` | + `ndjson` | `human` | | `start`, `send` | + `ndjson` | `json` / `ndjson` with `--wait` | | `wait` | `auto` \| `human` \| `ndjson` | `ndjson` | @@ -202,6 +230,10 @@ One JSON object per line: { "type": "done", "thread": {}, "latestAssistantMessage": {} } ``` +Successful `ask --format ndjson` streams the same thread events and ends with a `result` object. +`ask --format json` returns one object with `answer`, `threadId`, `turnId`, `created`, `dispatch`, +and `archive`. Human output writes progress to stderr and only the final answer to stdout. + ## Examples ```sh diff --git a/src/application/service.ts b/src/application/service.ts index 3b9e072..750ec62 100644 --- a/src/application/service.ts +++ b/src/application/service.ts @@ -227,6 +227,9 @@ export type T3ThreadApplicationService = { readonly getThreadMessages: ( input: GetThreadMessagesInput, ) => Effect.Effect<OrchestrationThreadDetailSnapshot, ApplicationError>; + readonly getThreadSummary: ( + threadId: string, + ) => Effect.Effect<OrchestrationThreadShell, ApplicationError>; readonly showThread: (threadId: string) => Effect.Effect<ThreadShow, ApplicationError>; readonly approveThread: (input: { readonly threadId: string; diff --git a/src/application/threads.ts b/src/application/threads.ts index a182fd8..5e2e48b 100644 --- a/src/application/threads.ts +++ b/src/application/threads.ts @@ -117,6 +117,23 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function ) { return yield* orchestration.getThreadDetailSnapshot(input); }); + const getThreadSummary = Effect.fn("T3ApplicationLive.getThreadSummary")(function* ( + threadId: string, + ) { + const snapshot = yield* loadThreadsSnapshot("all").pipe( + Effect.provideService(T3Orchestration, orchestration), + ); + const thread = snapshot.threads.find((item) => item.id === threadId); + if (thread === undefined) { + return yield* Effect.fail( + new ThreadLookupError({ + message: `thread not found: ${threadId}`, + threadId, + }), + ); + } + return thread; + }); const showThread = Effect.fn("T3ApplicationLive.showThread")(function* (threadId: string) { const thread = yield* orchestration.getThreadSnapshot(threadId); return projectThreadShow(thread); @@ -125,7 +142,9 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function const command = yield* makeThreadArchiveCommand(threadId).pipe( Effect.provideService(Crypto.Crypto, crypto), ); - return yield* orchestration.dispatch(command); + const dispatch = yield* orchestration.dispatch(command); + yield* awaitShellSequence(dispatch.sequence); + return dispatch; }); const unarchiveThread = Effect.fn("T3ApplicationLive.unarchiveThread")(function* ( threadId: string, @@ -365,6 +384,7 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function listThreads, searchThreads, getThreadMessages, + getThreadSummary, respondToThread, sendThread, showThread, diff --git a/src/cli/app.ts b/src/cli/app.ts index e8b7d42..e9d7a68 100644 --- a/src/cli/app.ts +++ b/src/cli/app.ts @@ -1,6 +1,7 @@ import { Command } from "effect/unstable/cli"; import { createActionCommand } from "./action.ts"; +import { askCommand } from "./ask.ts"; import { createAuthCommand } from "./auth.ts"; import { createEnvCommand } from "./env.ts"; import { cliEnvironmentSetting } from "./env/flag.ts"; @@ -22,6 +23,7 @@ export function createCliCommand() { Command.withGlobalFlags([cliEnvironmentSetting]), Command.withSubcommands([ createActionCommand(), + askCommand, createAuthCommand(), createEnvCommand(), listThreadsCommand, diff --git a/src/cli/ask-lifecycle.ts b/src/cli/ask-lifecycle.ts new file mode 100644 index 0000000..a6af6b3 --- /dev/null +++ b/src/cli/ask-lifecycle.ts @@ -0,0 +1,285 @@ +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; +import type { + OrchestrationMessage, + OrchestrationThread, + OrchestrationThreadShell, +} from "@t3tools/contracts"; + +import type { T3ApplicationService } from "../application/service.ts"; +import type { T3CliEnvShape } from "../config/env/env.ts"; +import { ThreadSessionError } from "../domain/error.ts"; +import { AskThreadArchivedError, AskThreadPendingRequestError } from "./error.ts"; +import { formatWaitEventNdjson } from "./format/thread.ts"; +import { isInteractiveHumanTerminal } from "./format/output.ts"; +import { T3Output } from "./output/service.ts"; +import { CliRuntime } from "./runtime/service.ts"; + +export const archivePolicyChoices = ["never", "always", "on-success", "on-failure"] as const; + +export type ArchivePolicy = (typeof archivePolicyChoices)[number]; +export type AskFormat = "human" | "json" | "ndjson"; + +export type AskArchiveResult = + | { + readonly policy: ArchivePolicy; + readonly status: "skipped"; + } + | { + readonly policy: ArchivePolicy; + readonly status: "archived"; + readonly sequence: number; + } + | { + readonly policy: ArchivePolicy; + readonly status: "failed"; + readonly error: string; + }; + +export interface AskExecutionState { + readonly created: boolean; + readonly archivePolicy: ArchivePolicy; + threadId: string | undefined; + dispatched: boolean; + baselineActiveTurnId: string | null; + archiveResult: AskArchiveResult | undefined; +} + +export function resolveAskFormat( + format: "auto" | AskFormat, + cliRuntime: CliRuntime["Service"], + t3CliEnv: T3CliEnvShape, +): AskFormat { + if (format !== "auto") { + return format; + } + return isInteractiveHumanTerminal(cliRuntime, t3CliEnv) ? "human" : "json"; +} + +export function ensureAskTargetAvailable(thread: OrchestrationThreadShell) { + if (thread.archivedAt !== null) { + return Effect.fail( + new AskThreadArchivedError({ + message: `thread is archived: ${thread.id}`, + threadId: thread.id, + }), + ); + } + if (thread.hasPendingApprovals || thread.hasPendingUserInput) { + return Effect.fail( + new AskThreadPendingRequestError({ + message: `thread has a pending approval or user-input request: ${thread.id}`, + threadId: thread.id, + }), + ); + } + return Effect.void; +} + +export function waitForBusyThread(application: T3ApplicationService, threadId: string) { + return application.watchThread(threadId).pipe( + Stream.tap((event) => + event.type === "status" ? ensureNoPendingRequest(application, threadId) : Effect.void, + ), + Stream.runLast, + Effect.flatMap((last) => { + const event = Option.getOrUndefined(last); + if (event?.type === "done") { + return Effect.void; + } + return Effect.fail( + new ThreadSessionError({ + message: `thread wait ended without a terminal event: ${threadId}`, + threadId, + }), + ); + }), + ); +} + +export function waitForAskThread( + application: T3ApplicationService, + output: T3Output["Service"], + input: { + readonly threadId: string; + readonly format: AskFormat; + }, +) { + let lastStatus = ""; + return Effect.gen(function* () { + if (input.format === "human") { + yield* output.writeStderr(`waiting for ${input.threadId}...\n`); + } + const last = yield* application.watchThread(input.threadId).pipe( + Stream.tap((event) => + Effect.gen(function* () { + if (event.type === "status") { + yield* ensureNoPendingRequest(application, input.threadId); + } + if (input.format === "ndjson") { + yield* output.printNdjson(formatWaitEventNdjson(event)); + return; + } + if (input.format === "human" && event.type === "status" && event.status !== lastStatus) { + lastStatus = event.status; + yield* output.writeStderr(`${input.threadId}: ${event.status}\n`); + } + }), + ), + Stream.runLast, + ); + const event = Option.getOrUndefined(last); + if (event?.type !== "done") { + return yield* Effect.fail( + new ThreadSessionError({ + message: `thread wait ended without a terminal event: ${input.threadId}`, + threadId: input.threadId, + }), + ); + } + return event.thread; + }); +} + +export function announceQueue(output: T3Output["Service"], format: AskFormat, threadId: string) { + if (format === "ndjson") { + return output.printNdjson({ type: "queue", status: "waiting", threadId }); + } + if (format === "human") { + return output.writeStderr(`thread ${threadId} is busy; waiting to ask...\n`); + } + return Effect.void; +} + +export function selectAskAnswer( + thread: OrchestrationThread, + baselineMessageIds: ReadonlySet<string>, +): OrchestrationMessage | undefined { + const newMessages = thread.messages.filter((message) => !baselineMessageIds.has(message.id)); + const userIndex = newMessages.findLastIndex((message) => message.role === "user"); + if (userIndex === -1) { + return undefined; + } + const turnId = thread.latestTurn?.turnId; + const candidates = newMessages + .slice(userIndex + 1) + .filter( + (message) => + message.role === "assistant" && + !message.streaming && + message.text.trim().length > 0 && + (turnId === undefined || message.turnId === turnId), + ); + return candidates.at(-1); +} + +export function finalizeArchive( + application: T3ApplicationService, + output: T3Output["Service"], + state: AskExecutionState, + succeeded: boolean, +): Effect.Effect<AskArchiveResult> { + if (state.archiveResult !== undefined) { + return Effect.succeed(state.archiveResult); + } + const shouldArchive = + state.archivePolicy === "always" || + (state.archivePolicy === "on-success" && succeeded) || + (state.archivePolicy === "on-failure" && !succeeded); + if (!state.dispatched || state.threadId === undefined || !shouldArchive) { + const result = { + policy: state.archivePolicy, + status: "skipped", + } satisfies AskArchiveResult; + state.archiveResult = result; + return Effect.succeed(result); + } + const threadId = state.threadId; + return application.archiveThread(threadId).pipe( + Effect.matchEffect({ + onFailure: (error) => + Effect.gen(function* () { + const result = { + policy: state.archivePolicy, + status: "failed", + error: error.message, + } satisfies AskArchiveResult; + state.archiveResult = result; + yield* output + .writeStderr(`warning: failed to archive thread ${threadId}: ${error.message}\n`) + .pipe(Effect.ignore); + return result; + }), + onSuccess: (dispatch) => { + const result = { + policy: state.archivePolicy, + status: "archived", + sequence: dispatch.sequence, + } satisfies AskArchiveResult; + state.archiveResult = result; + return Effect.succeed(result); + }, + }), + ); +} + +export function cleanupInterruptedAsk( + application: T3ApplicationService, + output: T3Output["Service"], + state: AskExecutionState, +) { + if (!state.dispatched || state.threadId === undefined) { + return Effect.void; + } + const threadId = state.threadId; + return Effect.gen(function* () { + let shouldInterrupt = state.created || state.baselineActiveTurnId === null; + if (!state.created && state.baselineActiveTurnId !== null) { + shouldInterrupt = yield* application.showThread(threadId).pipe( + Effect.matchEffect({ + onFailure: (error) => + output + .writeStderr( + `warning: could not inspect thread ${threadId} during cancellation: ${error.message}\n`, + ) + .pipe(Effect.ignore, Effect.as(false)), + onSuccess: (thread) => + Effect.succeed(thread.session?.activeTurnId !== state.baselineActiveTurnId), + }), + ); + } + if (shouldInterrupt) { + yield* application.interruptThread(threadId).pipe( + Effect.matchEffect({ + onFailure: (error) => + output + .writeStderr(`warning: failed to interrupt thread ${threadId}: ${error.message}\n`) + .pipe(Effect.ignore), + onSuccess: () => Effect.void, + }), + ); + } + yield* finalizeArchive(application, output, state, false); + }).pipe(Effect.asVoid); +} + +export function ensureTrailingNewline(text: string) { + return text.endsWith("\n") ? text : `${text}\n`; +} + +function ensureNoPendingRequest(application: T3ApplicationService, threadId: string) { + return application.showThread(threadId).pipe( + Effect.flatMap((thread) => { + if (!thread.hasPendingApprovals && !thread.hasPendingUserInput) { + return Effect.void; + } + return Effect.fail( + new AskThreadPendingRequestError({ + message: `thread requested approval or user input instead of answering: ${threadId}`, + threadId, + }), + ); + }), + ); +} diff --git a/src/cli/ask.ts b/src/cli/ask.ts new file mode 100644 index 0000000..76e7a58 --- /dev/null +++ b/src/cli/ask.ts @@ -0,0 +1,407 @@ +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import { Argument, Command, Flag } from "effect/unstable/cli"; +import type { DispatchResult } from "@t3tools/contracts"; + +import { T3Application } from "../application/service.ts"; +import { isThreadActive } from "../domain/thread-lifecycle.ts"; +import { ThreadSessionError } from "../domain/error.ts"; +import { loadT3CliEnv } from "../config/env/env.ts"; +import { + announceQueue, + archivePolicyChoices, + cleanupInterruptedAsk, + ensureAskTargetAvailable, + ensureTrailingNewline, + finalizeArchive, + resolveAskFormat, + selectAskAnswer, + waitForAskThread, + waitForBusyThread, +} from "./ask-lifecycle.ts"; +import type { AskExecutionState } from "./ask-lifecycle.ts"; +import { extraArgsConfig } from "./extra-args.ts"; +import { + AskNoAnswerError, + AskProjectMismatchError, + AskThreadBusyError, + AskTimeoutError, + InvalidAskTimeoutError, + InvalidFlagCombinationError, +} from "./error.ts"; +import { humanJsonNdjsonFormatChoices } from "./format/output.ts"; +import { modelFlags, projectFlag, selfActionForceFlag, worktreeFlag } from "./flags.ts"; +import { T3Input } from "./input/service.ts"; +import { requireSelfActionConfirmation } from "./interaction/self-action.ts"; +import { readInitialMessage } from "./message-input.ts"; +import { buildModelOptions } from "./model-options.ts"; +import { T3Output } from "./output/service.ts"; +import { requireCommandProjectRef } from "./require.ts"; +import { resolveWorktreePath } from "./scope/index.ts"; +import { CliRuntime } from "./runtime/service.ts"; + +const busyPolicyChoices = ["fail", "queue", "steer"] as const; + +interface AskResponse { + readonly answer: string; + readonly threadId: string; + readonly turnId: string | null; + readonly created: boolean; + readonly dispatch: DispatchResult; +} + +const askThreadFlag = Flag.string("thread").pipe( + Flag.withDescription("Existing thread id; T3CODE_THREAD_ID is not used"), + Flag.optional, +); + +const archivePolicyFlag = Flag.choice("archive", archivePolicyChoices).pipe( + Flag.withDescription( + "Archive policy (default: on-success for created threads, never for existing threads)", + ), + Flag.optional, +); + +const busyPolicyFlag = Flag.choice("busy", busyPolicyChoices).pipe( + Flag.withDescription("Existing-thread busy policy (default: fail)"), + Flag.optional, +); + +const timeoutFlag = Flag.string("timeout").pipe( + Flag.withDescription("Queue and response timeout, such as 30s, 5m, or 1h"), + Flag.optional, +); + +const askFormatFlag = Flag.choice("format", humanJsonNdjsonFormatChoices).pipe( + Flag.withDefault("human"), +); + +export const askCommand = Command.make( + "ask", + { + project: projectFlag, + thread: askThreadFlag, + force: selfActionForceFlag, + message: Argument.string("message").pipe(Argument.optional), + stdin: Flag.boolean("stdin"), + title: Flag.string("title").pipe(Flag.optional), + worktree: worktreeFlag, + provider: Flag.string("provider").pipe(Flag.optional), + model: Flag.string("model").pipe(Flag.optional), + ...modelFlags, + archive: archivePolicyFlag, + busy: busyPolicyFlag, + timeout: timeoutFlag, + format: askFormatFlag, + ...extraArgsConfig, + }, + ({ + project, + thread, + force, + message, + stdin, + title, + worktree, + provider, + model, + option, + reasoningEffort, + effort, + fastMode, + thinking, + archive, + busy, + timeout, + format, + }) => + Effect.gen(function* () { + const explicitThreadId = Option.getOrUndefined(thread); + const created = explicitThreadId === undefined; + const explicitBusyPolicy = Option.getOrUndefined(busy); + const titleValue = Option.getOrUndefined(title); + const providerValue = Option.getOrUndefined(provider); + const modelValue = Option.getOrUndefined(model); + const createOnlyFlags = [ + Option.isSome(title) ? "--title" : null, + Option.isSome(worktree) ? "--worktree" : null, + Option.isSome(provider) ? "--provider" : null, + Option.isSome(model) ? "--model" : null, + ].filter((flag): flag is string => flag !== null); + + if (!created && createOnlyFlags.length > 0) { + return yield* Effect.fail( + new InvalidFlagCombinationError({ + message: `${createOnlyFlags.join(", ")} cannot be used with --thread`, + }), + ); + } + if (created && explicitBusyPolicy !== undefined) { + return yield* Effect.fail( + new InvalidFlagCombinationError({ + message: "--busy can only be used with --thread", + }), + ); + } + if (created && force) { + return yield* Effect.fail( + new InvalidFlagCombinationError({ + message: "--force can only be used with --thread", + }), + ); + } + + const timeoutValue = Option.getOrUndefined(timeout); + const timeoutDuration = + timeoutValue === undefined ? undefined : yield* parseAskTimeout(timeoutValue); + const archivePolicy = Option.getOrElse(archive, () => (created ? "on-success" : "never")); + const busyPolicy = explicitBusyPolicy ?? "fail"; + const application = yield* T3Application; + const cliRuntime = yield* CliRuntime; + const t3CliEnv = yield* loadT3CliEnv; + const input = yield* T3Input; + const output = yield* T3Output; + const resolvedFormat = resolveAskFormat(format, cliRuntime, t3CliEnv); + const options = buildModelOptions({ + option, + reasoningEffort, + effort, + fastMode, + thinking, + }); + const state: AskExecutionState = { + created, + archivePolicy, + threadId: explicitThreadId, + dispatched: false, + baselineActiveTurnId: null, + archiveResult: undefined, + }; + + const response = Effect.gen(function* () { + const text = yield* readInitialMessage({ + message: Option.getOrUndefined(message), + fromStdin: stdin, + readStdin: input.readStdin, + }); + let baselineMessageIds = new Set<string>(); + let dispatch: DispatchResult; + + if (explicitThreadId === undefined) { + const projectRef = yield* requireCommandProjectRef({ project }); + const resolvedProject = yield* application.resolveProject(projectRef); + const worktreePath = resolveWorktreePath({ + value: Option.getOrUndefined(worktree), + scope: t3CliEnv.scope, + }); + const result = yield* application.startThread( + { + message: text, + projectRef: resolvedProject.id, + ...(titleValue !== undefined ? { title: titleValue } : {}), + ...(worktreePath !== undefined ? { worktreePath } : {}), + ...(providerValue !== undefined ? { provider: providerValue } : {}), + ...(modelValue !== undefined ? { model: modelValue } : {}), + ...(options.length > 0 ? { options } : {}), + }, + { until: "dispatch" }, + ); + state.threadId = result.threadId; + state.dispatched = true; + dispatch = result.dispatch; + } else { + let summary = yield* application.getThreadSummary(explicitThreadId); + yield* ensureAskTargetAvailable(summary); + + const projectRef = Option.getOrUndefined(project); + if (projectRef !== undefined) { + const resolvedProject = yield* application.resolveProject(projectRef); + if (summary.projectId !== resolvedProject.id) { + return yield* Effect.fail( + new AskProjectMismatchError({ + message: `thread ${explicitThreadId} belongs to project ${summary.projectId}, not ${resolvedProject.id}`, + threadId: explicitThreadId, + projectId: resolvedProject.id, + }), + ); + } + } + + yield* requireSelfActionConfirmation({ + threadId: explicitThreadId, + force, + cliRuntime, + t3CliEnv, + action: "ask", + }); + + if (isThreadActive(summary)) { + if (busyPolicy === "fail") { + return yield* Effect.fail( + new AskThreadBusyError({ + message: `thread is busy: ${explicitThreadId}; use --busy queue or --busy steer`, + threadId: explicitThreadId, + }), + ); + } + if (busyPolicy === "queue") { + yield* announceQueue(output, resolvedFormat, explicitThreadId); + while (isThreadActive(summary)) { + yield* waitForBusyThread(application, explicitThreadId); + summary = yield* application.getThreadSummary(explicitThreadId); + yield* ensureAskTargetAvailable(summary); + } + } + } + + const baseline = yield* application.getThreadMessages({ threadId: explicitThreadId }); + baselineMessageIds = new Set(baseline.thread.messages.map((item) => item.id)); + state.baselineActiveTurnId = summary.session?.activeTurnId ?? null; + const result = yield* application.sendThread( + { + threadId: explicitThreadId, + message: text, + ...(options.length > 0 ? { options } : {}), + }, + { until: "dispatch" }, + ); + state.dispatched = true; + dispatch = result.dispatch; + } + + const threadId = state.threadId; + if (threadId === undefined) { + return yield* Effect.fail( + new AskNoAnswerError({ + message: "ask did not resolve a thread id", + threadId: "unknown", + }), + ); + } + + if (resolvedFormat === "ndjson") { + yield* output.printNdjson({ type: "dispatch", sequence: dispatch.sequence }); + } + yield* waitForAskThread(application, output, { + threadId, + format: resolvedFormat, + }); + + const finalSnapshot = yield* application.getThreadMessages({ threadId }); + if (finalSnapshot.thread.session?.status === "error") { + return yield* Effect.fail( + new ThreadSessionError({ + threadId, + message: finalSnapshot.thread.session.lastError ?? "thread ended with error", + }), + ); + } + const answer = selectAskAnswer(finalSnapshot.thread, baselineMessageIds); + if (answer === undefined) { + return yield* Effect.fail( + new AskNoAnswerError({ + message: `thread completed without a new final answer: ${threadId}`, + threadId, + }), + ); + } + return { + answer: answer.text, + threadId, + turnId: answer.turnId, + created, + dispatch, + } satisfies AskResponse; + }); + + const responseWithCleanup = response.pipe( + Effect.onInterrupt(() => cleanupInterruptedAsk(application, output, state)), + ); + const timedResponse = + timeoutDuration === undefined + ? responseWithCleanup + : responseWithCleanup.pipe( + Effect.timeoutOrElse({ + duration: timeoutDuration, + orElse: () => + Effect.fail( + new AskTimeoutError({ + message: `ask timed out after ${timeoutValue ?? Duration.format(timeoutDuration)}`, + timeout: timeoutValue ?? Duration.format(timeoutDuration), + }), + ), + }), + ); + + return yield* timedResponse.pipe( + Effect.matchEffect({ + onFailure: (error) => + finalizeArchive(application, output, state, false).pipe( + Effect.flatMap(() => Effect.fail(error)), + ), + onSuccess: (result) => + Effect.gen(function* () { + const archiveResult = yield* finalizeArchive(application, output, state, true); + const formatted = { + answer: result.answer, + threadId: result.threadId, + turnId: result.turnId, + created: result.created, + dispatch: result.dispatch, + archive: archiveResult, + }; + if (resolvedFormat === "json") { + return yield* output.printJson(formatted); + } + if (resolvedFormat === "ndjson") { + return yield* output.printNdjson({ type: "result", ...formatted }); + } + return yield* output.writeStdout(ensureTrailingNewline(result.answer)); + }), + }), + ); + }), +).pipe(Command.withDescription("ask a project or existing thread and wait for one answer")); + +function parseAskTimeout(value: string) { + const trimmed = value.trim(); + const shorthand = /^([0-9]+(?:\.[0-9]+)?)(ms|s|m|h|d|w)$/.exec(trimmed); + let input = trimmed; + if (shorthand !== null) { + const amount = shorthand[1]; + const unit = shorthand[2]; + const units: Readonly<Record<string, string>> = { + ms: "millis", + s: "seconds", + m: "minutes", + h: "hours", + d: "days", + w: "weeks", + }; + const expandedUnit = unit === undefined ? undefined : units[unit]; + if (amount !== undefined && expandedUnit !== undefined) { + input = `${amount} ${expandedUnit}`; + } + } + const parsed = Schema.decodeUnknownOption(Schema.DurationFromString)(input); + if (Option.isNone(parsed)) { + return Effect.fail( + new InvalidAskTimeoutError({ + message: `invalid timeout: ${value}`, + value, + }), + ); + } + const millis = Duration.toMillis(parsed.value); + if (!Number.isFinite(millis) || millis <= 0) { + return Effect.fail( + new InvalidAskTimeoutError({ + message: `timeout must be a positive finite duration: ${value}`, + value, + }), + ); + } + return Effect.succeed(parsed.value); +} diff --git a/src/cli/error.ts b/src/cli/error.ts index 1a01f0e..500065f 100644 --- a/src/cli/error.ts +++ b/src/cli/error.ts @@ -39,6 +39,60 @@ export class InvalidFlagCombinationError extends Schema.TaggedErrorClass<Invalid }, ) {} +export class InvalidAskTimeoutError extends Schema.TaggedErrorClass<InvalidAskTimeoutError>()( + "InvalidAskTimeoutError", + { + message: Schema.String, + value: Schema.String, + }, +) {} + +export class AskThreadArchivedError extends Schema.TaggedErrorClass<AskThreadArchivedError>()( + "AskThreadArchivedError", + { + message: Schema.String, + threadId: Schema.String, + }, +) {} + +export class AskThreadBusyError extends Schema.TaggedErrorClass<AskThreadBusyError>()( + "AskThreadBusyError", + { + message: Schema.String, + threadId: Schema.String, + }, +) {} + +export class AskThreadPendingRequestError extends Schema.TaggedErrorClass<AskThreadPendingRequestError>()( + "AskThreadPendingRequestError", + { + message: Schema.String, + threadId: Schema.String, + }, +) {} + +export class AskProjectMismatchError extends Schema.TaggedErrorClass<AskProjectMismatchError>()( + "AskProjectMismatchError", + { + message: Schema.String, + threadId: Schema.String, + projectId: Schema.String, + }, +) {} + +export class AskNoAnswerError extends Schema.TaggedErrorClass<AskNoAnswerError>()( + "AskNoAnswerError", + { + message: Schema.String, + threadId: Schema.String, + }, +) {} + +export class AskTimeoutError extends Schema.TaggedErrorClass<AskTimeoutError>()("AskTimeoutError", { + message: Schema.String, + timeout: Schema.String, +}) {} + export class MissingRequestError extends Schema.TaggedErrorClass<MissingRequestError>()( "MissingRequestError", { From c757e237b220c5d9b683b1ba0923c9a635c9a423 Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 09:46:28 +0300 Subject: [PATCH 02/11] fix: wait for ask turn visibility --- flake.lock | 8 ++++---- flake.nix | 2 +- src/application/service.ts | 1 + src/application/threads.ts | 1 + src/cli/ask.ts | 1 + 5 files changed, 8 insertions(+), 5 deletions(-) diff --git a/flake.lock b/flake.lock index b78a9e4..547b4a6 100644 --- a/flake.lock +++ b/flake.lock @@ -42,17 +42,17 @@ "upstream-t3code": { "flake": false, "locked": { - "lastModified": 1786361740, - "narHash": "sha256-qZi9hMGzqpmnpqvvVtsQvkZIiVqTgOMWv1y15MiSAYg=", + "lastModified": 1786783355, + "narHash": "sha256-G6sttzF/SChyPCzR7nFIEMFnNjrBul0bP4pa7kS2M9M=", "owner": "pingdotgg", "repo": "t3code", - "rev": "3b72d17cbca691f0b64e6d4a10c9e349f42873a5", + "rev": "f0ebc628c6dd83fd0c7963078ad7778ce6028d0c", "type": "github" }, "original": { "owner": "pingdotgg", "repo": "t3code", - "rev": "3b72d17cbca691f0b64e6d4a10c9e349f42873a5", + "rev": "f0ebc628c6dd83fd0c7963078ad7778ce6028d0c", "type": "github" } } diff --git a/flake.nix b/flake.nix index b15ce7a..3a080b6 100644 --- a/flake.nix +++ b/flake.nix @@ -6,7 +6,7 @@ nixpkgs-darwin.url = "github:NixOS/nixpkgs/nixpkgs-26.05-darwin"; upstream-t3code = { - url = "github:pingdotgg/t3code/3b72d17cbca691f0b64e6d4a10c9e349f42873a5"; + url = "github:pingdotgg/t3code/f0ebc628c6dd83fd0c7963078ad7778ce6028d0c"; flake = false; }; }; diff --git a/src/application/service.ts b/src/application/service.ts index 750ec62..2f964c8 100644 --- a/src/application/service.ts +++ b/src/application/service.ts @@ -209,6 +209,7 @@ export class T3ProjectApplication extends Context.Service< >()("t3cli/T3ProjectApplication") {} export type T3ThreadApplicationService = { + readonly awaitShellSequence: (sequence: number) => Effect.Effect<void, ApplicationError>; readonly searchThreads: ( input: OrchestrationSearchThreadsInput, ) => Effect.Effect<ReadonlyArray<ThreadSearchResult>, ApplicationError>; diff --git a/src/application/threads.ts b/src/application/threads.ts index 5e2e48b..85c66e2 100644 --- a/src/application/threads.ts +++ b/src/application/threads.ts @@ -371,6 +371,7 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function return { approveThread, archiveThread, + awaitShellSequence, deleteThread, interruptThread, pinThread, diff --git a/src/cli/ask.ts b/src/cli/ask.ts index 76e7a58..fdef793 100644 --- a/src/cli/ask.ts +++ b/src/cli/ask.ts @@ -284,6 +284,7 @@ export const askCommand = Command.make( if (resolvedFormat === "ndjson") { yield* output.printNdjson({ type: "dispatch", sequence: dispatch.sequence }); } + yield* application.awaitShellSequence(dispatch.sequence); yield* waitForAskThread(application, output, { threadId, format: resolvedFormat, From c5ceaa0354355a61e53e22e94eaff4865b9b66a0 Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 10:05:07 +0300 Subject: [PATCH 03/11] fix: guard ask interruption ownership --- src/application/service.ts | 4 ++++ src/application/threads.ts | 14 ++++++++++++++ src/cli/ask-lifecycle.ts | 20 +++----------------- src/cli/ask.ts | 9 +++++++++ 4 files changed, 30 insertions(+), 17 deletions(-) diff --git a/src/application/service.ts b/src/application/service.ts index 2f964c8..c5a4c5a 100644 --- a/src/application/service.ts +++ b/src/application/service.ts @@ -250,6 +250,10 @@ export type T3ThreadApplicationService = { >; readonly archiveThread: (threadId: string) => Effect.Effect<DispatchResult, ApplicationError>; readonly interruptThread: (threadId: string) => Effect.Effect<DispatchResult, ApplicationError>; + readonly interruptThreadTurn: ( + threadId: string, + turnId: string, + ) => Effect.Effect<DispatchResult | undefined, ApplicationError>; readonly pinThread: (threadId: string) => Effect.Effect<DispatchResult, ApplicationError>; readonly settleThread: (threadId: string) => Effect.Effect<DispatchResult, ApplicationError>; readonly snoozeThread: ( diff --git a/src/application/threads.ts b/src/application/threads.ts index 85c66e2..f2fcb19 100644 --- a/src/application/threads.ts +++ b/src/application/threads.ts @@ -207,6 +207,19 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function }).pipe(Effect.provideService(Crypto.Crypto, crypto)); return yield* orchestration.dispatch(command); }); + const interruptThreadTurn = Effect.fn("T3ApplicationLive.interruptThreadTurn")(function* ( + threadId: string, + turnId: string, + ) { + const snapshot = yield* orchestration.getThreadSnapshot(threadId); + if (snapshot.session?.activeTurnId !== turnId) { + return undefined; + } + const command = yield* makeThreadInterruptCommand({ threadId, turnId }).pipe( + Effect.provideService(Crypto.Crypto, crypto), + ); + return yield* orchestration.dispatch(command); + }); const deleteThread = Effect.fn("T3ApplicationLive.deleteThread")(function* (threadId: string) { const snapshot = yield* loadThreadsSnapshot("all").pipe( Effect.provideService(T3Orchestration, orchestration), @@ -374,6 +387,7 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function awaitShellSequence, deleteThread, interruptThread, + interruptThreadTurn, pinThread, settleThread, snoozeThread, diff --git a/src/cli/ask-lifecycle.ts b/src/cli/ask-lifecycle.ts index a6af6b3..d53cfb1 100644 --- a/src/cli/ask-lifecycle.ts +++ b/src/cli/ask-lifecycle.ts @@ -43,6 +43,7 @@ export interface AskExecutionState { threadId: string | undefined; dispatched: boolean; baselineActiveTurnId: string | null; + askTurnId: string | null; archiveResult: AskArchiveResult | undefined; } @@ -234,23 +235,8 @@ export function cleanupInterruptedAsk( } const threadId = state.threadId; return Effect.gen(function* () { - let shouldInterrupt = state.created || state.baselineActiveTurnId === null; - if (!state.created && state.baselineActiveTurnId !== null) { - shouldInterrupt = yield* application.showThread(threadId).pipe( - Effect.matchEffect({ - onFailure: (error) => - output - .writeStderr( - `warning: could not inspect thread ${threadId} during cancellation: ${error.message}\n`, - ) - .pipe(Effect.ignore, Effect.as(false)), - onSuccess: (thread) => - Effect.succeed(thread.session?.activeTurnId !== state.baselineActiveTurnId), - }), - ); - } - if (shouldInterrupt) { - yield* application.interruptThread(threadId).pipe( + if (state.askTurnId !== null) { + yield* application.interruptThreadTurn(threadId, state.askTurnId).pipe( Effect.matchEffect({ onFailure: (error) => output diff --git a/src/cli/ask.ts b/src/cli/ask.ts index fdef793..1020981 100644 --- a/src/cli/ask.ts +++ b/src/cli/ask.ts @@ -177,6 +177,7 @@ export const askCommand = Command.make( threadId: explicitThreadId, dispatched: false, baselineActiveTurnId: null, + askTurnId: null, archiveResult: undefined, }; @@ -285,6 +286,14 @@ export const askCommand = Command.make( yield* output.printNdjson({ type: "dispatch", sequence: dispatch.sequence }); } yield* application.awaitShellSequence(dispatch.sequence); + const visibleSummary = yield* application.getThreadSummary(threadId); + const visibleActiveTurnId = visibleSummary.session?.activeTurnId ?? null; + if ( + visibleActiveTurnId !== null && + (state.created || visibleActiveTurnId !== state.baselineActiveTurnId) + ) { + state.askTurnId = visibleActiveTurnId; + } yield* waitForAskThread(application, output, { threadId, format: resolvedFormat, From 3fabd0c7ef55830a3d5b6a18ce94ff0b3db8a7ed Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 10:24:56 +0300 Subject: [PATCH 04/11] fix: bind ask answer to dispatched message --- src/application/service.ts | 2 ++ src/application/threads.ts | 14 +++++---- src/cli/ask-lifecycle.ts | 61 ++++++++++++++++++++++++++++++-------- src/cli/ask.ts | 10 ++++--- src/cli/threads/send.ts | 6 +++- src/cli/threads/start.ts | 7 ++++- 6 files changed, 75 insertions(+), 25 deletions(-) diff --git a/src/application/service.ts b/src/application/service.ts index c5a4c5a..9792f95 100644 --- a/src/application/service.ts +++ b/src/application/service.ts @@ -278,6 +278,7 @@ export type T3ThreadApplicationService = { ) => Effect.Effect< { readonly dispatch: DispatchResult; + readonly messageId: string; readonly project: OrchestrationProjectShell; readonly threadId: string; readonly thread?: OrchestrationThread; @@ -290,6 +291,7 @@ export type T3ThreadApplicationService = { ) => Effect.Effect< { readonly dispatch: DispatchResult; + readonly messageId: string; readonly threadId: string; readonly thread?: OrchestrationThread; }, diff --git a/src/application/threads.ts b/src/application/threads.ts index f2fcb19..0dd14a9 100644 --- a/src/application/threads.ts +++ b/src/application/threads.ts @@ -288,9 +288,10 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function yield* awaitShellSequence(createDispatch.sequence); const dispatch = yield* orchestration.dispatch(commands.turnCommand); const threadId = commands.threadId; + const messageId = commands.turnCommand.message.messageId; const until = policy?.until ?? "dispatch"; if (until === "dispatch") { - return { dispatch, project: scope.project, threadId }; + return { dispatch, messageId, project: scope.project, threadId }; } yield* awaitShellSequence(dispatch.sequence); if (until === "visible") { @@ -300,11 +301,11 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function return opened.snapshot; }), ); - return { dispatch, project: scope.project, threadId, thread }; + return { dispatch, messageId, project: scope.project, threadId, thread }; } const thread = yield* awaitThreadCompletion(threadId); yield* failIfThreadError(thread); - return { dispatch, project: scope.project, threadId, thread }; + return { dispatch, messageId, project: scope.project, threadId, thread }; }); const sendThread = Effect.fn("T3ApplicationLive.sendThread")(function* ( input: SendThreadInput, @@ -324,9 +325,10 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function ...(modelSelection !== undefined ? { modelSelection } : {}), }).pipe(Effect.provideService(Crypto.Crypto, crypto)); const dispatch = yield* orchestration.dispatch(command); + const messageId = command.message.messageId; const until = policy?.until ?? "dispatch"; if (until === "dispatch") { - return { dispatch, threadId: input.threadId }; + return { dispatch, messageId, threadId: input.threadId }; } yield* awaitShellSequence(dispatch.sequence); if (until === "visible") { @@ -336,11 +338,11 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function return opened.snapshot; }), ); - return { dispatch, threadId: input.threadId, thread }; + return { dispatch, messageId, threadId: input.threadId, thread }; } const thread = yield* awaitThreadCompletion(input.threadId); yield* failIfThreadError(thread); - return { dispatch, threadId: input.threadId, thread }; + return { dispatch, messageId, threadId: input.threadId, thread }; }); const watchThread = (threadId: string) => streamThreadEvents(threadId); const waitForThread = Effect.fn("T3ApplicationLive.waitForThread")(function* (threadId: string) { diff --git a/src/cli/ask-lifecycle.ts b/src/cli/ask-lifecycle.ts index d53cfb1..72f8bce 100644 --- a/src/cli/ask-lifecycle.ts +++ b/src/cli/ask-lifecycle.ts @@ -105,9 +105,12 @@ export function waitForAskThread( input: { readonly threadId: string; readonly format: AskFormat; + readonly messageId: string; + readonly state: AskExecutionState; }, ) { let lastStatus = ""; + let askWindowOpen = false; return Effect.gen(function* () { if (input.format === "human") { yield* output.writeStderr(`waiting for ${input.threadId}...\n`); @@ -118,6 +121,36 @@ export function waitForAskThread( if (event.type === "status") { yield* ensureNoPendingRequest(application, input.threadId); } + if (event.type === "thread") { + const userIndex = event.thread.messages.findIndex( + (message) => message.id === input.messageId && message.role === "user", + ); + if (userIndex !== -1) { + const following = event.thread.messages.slice(userIndex + 1); + const nextUserIndex = following.findIndex((message) => message.role === "user"); + const askMessages = + nextUserIndex === -1 ? following : following.slice(0, nextUserIndex); + const turnId = askMessages.findLast( + (message) => message.role === "assistant" && message.turnId !== null, + )?.turnId; + if (turnId !== undefined && turnId !== null) { + input.state.askTurnId = turnId; + } + askWindowOpen = nextUserIndex === -1; + } + } else if (event.type === "message") { + if (event.message.id === input.messageId) { + askWindowOpen = true; + } else if (askWindowOpen && event.message.role === "user") { + askWindowOpen = false; + } else if ( + askWindowOpen && + event.message.role === "assistant" && + event.message.turnId !== null + ) { + input.state.askTurnId = event.message.turnId; + } + } if (input.format === "ndjson") { yield* output.printNdjson(formatWaitEventNdjson(event)); return; @@ -155,23 +188,25 @@ export function announceQueue(output: T3Output["Service"], format: AskFormat, th export function selectAskAnswer( thread: OrchestrationThread, - baselineMessageIds: ReadonlySet<string>, + messageId: string, + turnId: string | null, ): OrchestrationMessage | undefined { - const newMessages = thread.messages.filter((message) => !baselineMessageIds.has(message.id)); - const userIndex = newMessages.findLastIndex((message) => message.role === "user"); + const userIndex = thread.messages.findIndex( + (message) => message.id === messageId && message.role === "user", + ); if (userIndex === -1) { return undefined; } - const turnId = thread.latestTurn?.turnId; - const candidates = newMessages - .slice(userIndex + 1) - .filter( - (message) => - message.role === "assistant" && - !message.streaming && - message.text.trim().length > 0 && - (turnId === undefined || message.turnId === turnId), - ); + const following = thread.messages.slice(userIndex + 1); + const nextUserIndex = following.findIndex((message) => message.role === "user"); + const askMessages = nextUserIndex === -1 ? following : following.slice(0, nextUserIndex); + const candidates = askMessages.filter( + (message) => + message.role === "assistant" && + !message.streaming && + message.text.trim().length > 0 && + (turnId === null || message.turnId === turnId), + ); return candidates.at(-1); } diff --git a/src/cli/ask.ts b/src/cli/ask.ts index 1020981..b4c1c59 100644 --- a/src/cli/ask.ts +++ b/src/cli/ask.ts @@ -187,8 +187,8 @@ export const askCommand = Command.make( fromStdin: stdin, readStdin: input.readStdin, }); - let baselineMessageIds = new Set<string>(); let dispatch: DispatchResult; + let askMessageId: string; if (explicitThreadId === undefined) { const projectRef = yield* requireCommandProjectRef({ project }); @@ -212,6 +212,7 @@ export const askCommand = Command.make( state.threadId = result.threadId; state.dispatched = true; dispatch = result.dispatch; + askMessageId = result.messageId; } else { let summary = yield* application.getThreadSummary(explicitThreadId); yield* ensureAskTargetAvailable(summary); @@ -257,8 +258,6 @@ export const askCommand = Command.make( } } - const baseline = yield* application.getThreadMessages({ threadId: explicitThreadId }); - baselineMessageIds = new Set(baseline.thread.messages.map((item) => item.id)); state.baselineActiveTurnId = summary.session?.activeTurnId ?? null; const result = yield* application.sendThread( { @@ -270,6 +269,7 @@ export const askCommand = Command.make( ); state.dispatched = true; dispatch = result.dispatch; + askMessageId = result.messageId; } const threadId = state.threadId; @@ -297,6 +297,8 @@ export const askCommand = Command.make( yield* waitForAskThread(application, output, { threadId, format: resolvedFormat, + messageId: askMessageId, + state, }); const finalSnapshot = yield* application.getThreadMessages({ threadId }); @@ -308,7 +310,7 @@ export const askCommand = Command.make( }), ); } - const answer = selectAskAnswer(finalSnapshot.thread, baselineMessageIds); + const answer = selectAskAnswer(finalSnapshot.thread, askMessageId, state.askTurnId); if (answer === undefined) { return yield* Effect.fail( new AskNoAnswerError({ diff --git a/src/cli/threads/send.ts b/src/cli/threads/send.ts index 884925d..c29a939 100644 --- a/src/cli/threads/send.ts +++ b/src/cli/threads/send.ts @@ -118,7 +118,11 @@ export const sendThreadCommand = Command.make( const result = yield* application.sendThread(input, { until: "visible" }); if (resolvedFormat === "json") { - return yield* output.printJson(result); + return yield* output.printJson({ + dispatch: result.dispatch, + threadId: result.threadId, + thread: result.thread, + }); } return yield* output.printInfo(`message sent: ${result.threadId}`); }), diff --git a/src/cli/threads/start.ts b/src/cli/threads/start.ts index b57f6a1..098df08 100644 --- a/src/cli/threads/start.ts +++ b/src/cli/threads/start.ts @@ -129,7 +129,12 @@ export const startThreadCommand = Command.make( const result = yield* application.startThread(input, { until: "visible" }); if (resolvedFormat === "json") { - yield* output.printJson(result); + yield* output.printJson({ + dispatch: result.dispatch, + project: result.project, + threadId: result.threadId, + thread: result.thread, + }); } else { yield* output.printInfo( formatThreadStartedHuman({ From 36762cda7f8c9fee8aa3d5aeb7d1b5b1aa90bb5a Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 10:32:05 +0300 Subject: [PATCH 05/11] fix: prefer completed ask answer --- src/cli/ask.ts | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/cli/ask.ts b/src/cli/ask.ts index b4c1c59..ba79b99 100644 --- a/src/cli/ask.ts +++ b/src/cli/ask.ts @@ -302,16 +302,16 @@ export const askCommand = Command.make( }); const finalSnapshot = yield* application.getThreadMessages({ threadId }); - if (finalSnapshot.thread.session?.status === "error") { - return yield* Effect.fail( - new ThreadSessionError({ - threadId, - message: finalSnapshot.thread.session.lastError ?? "thread ended with error", - }), - ); - } const answer = selectAskAnswer(finalSnapshot.thread, askMessageId, state.askTurnId); if (answer === undefined) { + if (finalSnapshot.thread.session?.status === "error") { + return yield* Effect.fail( + new ThreadSessionError({ + threadId, + message: finalSnapshot.thread.session.lastError ?? "thread ended with error", + }), + ); + } return yield* Effect.fail( new AskNoAnswerError({ message: `thread completed without a new final answer: ${threadId}`, From c6c64e434678a8f027c7fd87790dacac0bdb5688 Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 10:43:44 +0300 Subject: [PATCH 06/11] fix: scope ask lifecycle to dispatched message --- src/cli/ask-lifecycle.ts | 29 ++++++++++++++++++++++++----- src/cli/ask.ts | 11 ----------- 2 files changed, 24 insertions(+), 16 deletions(-) diff --git a/src/cli/ask-lifecycle.ts b/src/cli/ask-lifecycle.ts index 72f8bce..d384270 100644 --- a/src/cli/ask-lifecycle.ts +++ b/src/cli/ask-lifecycle.ts @@ -38,11 +38,9 @@ export type AskArchiveResult = }; export interface AskExecutionState { - readonly created: boolean; readonly archivePolicy: ArchivePolicy; threadId: string | undefined; dispatched: boolean; - baselineActiveTurnId: string | null; askTurnId: string | null; archiveResult: AskArchiveResult | undefined; } @@ -111,6 +109,7 @@ export function waitForAskThread( ) { let lastStatus = ""; let askWindowOpen = false; + let answerFound = false; return Effect.gen(function* () { if (input.format === "human") { yield* output.writeStderr(`waiting for ${input.threadId}...\n`); @@ -122,6 +121,13 @@ export function waitForAskThread( yield* ensureNoPendingRequest(application, input.threadId); } if (event.type === "thread") { + const answer = selectAskAnswer(event.thread, input.messageId, input.state.askTurnId); + if (answer !== undefined) { + if (answer.turnId !== null) { + input.state.askTurnId = answer.turnId; + } + answerFound = true; + } const userIndex = event.thread.messages.findIndex( (message) => message.id === input.messageId && message.role === "user", ); @@ -148,7 +154,19 @@ export function waitForAskThread( event.message.role === "assistant" && event.message.turnId !== null ) { - input.state.askTurnId = event.message.turnId; + if ( + input.state.askTurnId === null || + input.state.askTurnId === event.message.turnId + ) { + input.state.askTurnId = event.message.turnId; + answerFound = !event.message.streaming && event.message.text.trim().length > 0; + } + } else if ( + askWindowOpen && + event.message.role === "assistant" && + input.state.askTurnId === null + ) { + answerFound = !event.message.streaming && event.message.text.trim().length > 0; } } if (input.format === "ndjson") { @@ -161,10 +179,11 @@ export function waitForAskThread( } }), ), + Stream.takeUntil((event) => answerFound || event.type === "done"), Stream.runLast, ); const event = Option.getOrUndefined(last); - if (event?.type !== "done") { + if (!answerFound && event?.type !== "done") { return yield* Effect.fail( new ThreadSessionError({ message: `thread wait ended without a terminal event: ${input.threadId}`, @@ -172,7 +191,7 @@ export function waitForAskThread( }), ); } - return event.thread; + return yield* Effect.void; }); } diff --git a/src/cli/ask.ts b/src/cli/ask.ts index ba79b99..59a538a 100644 --- a/src/cli/ask.ts +++ b/src/cli/ask.ts @@ -172,11 +172,9 @@ export const askCommand = Command.make( thinking, }); const state: AskExecutionState = { - created, archivePolicy, threadId: explicitThreadId, dispatched: false, - baselineActiveTurnId: null, askTurnId: null, archiveResult: undefined, }; @@ -258,7 +256,6 @@ export const askCommand = Command.make( } } - state.baselineActiveTurnId = summary.session?.activeTurnId ?? null; const result = yield* application.sendThread( { threadId: explicitThreadId, @@ -286,14 +283,6 @@ export const askCommand = Command.make( yield* output.printNdjson({ type: "dispatch", sequence: dispatch.sequence }); } yield* application.awaitShellSequence(dispatch.sequence); - const visibleSummary = yield* application.getThreadSummary(threadId); - const visibleActiveTurnId = visibleSummary.session?.activeTurnId ?? null; - if ( - visibleActiveTurnId !== null && - (state.created || visibleActiveTurnId !== state.baselineActiveTurnId) - ) { - state.askTurnId = visibleActiveTurnId; - } yield* waitForAskThread(application, output, { threadId, format: resolvedFormat, From 95172e25a4f41bc5804d6fb4a120b08758c2854e Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 10:53:05 +0300 Subject: [PATCH 07/11] fix: use full thread state for busy checks --- src/cli/ask.ts | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/cli/ask.ts b/src/cli/ask.ts index 59a538a..214954f 100644 --- a/src/cli/ask.ts +++ b/src/cli/ask.ts @@ -237,7 +237,9 @@ export const askCommand = Command.make( action: "ask", }); - if (isThreadActive(summary)) { + let targetThread = (yield* application.getThreadMessages({ threadId: explicitThreadId })) + .thread; + if (isThreadActive(targetThread)) { if (busyPolicy === "fail") { return yield* Effect.fail( new AskThreadBusyError({ @@ -248,10 +250,13 @@ export const askCommand = Command.make( } if (busyPolicy === "queue") { yield* announceQueue(output, resolvedFormat, explicitThreadId); - while (isThreadActive(summary)) { + while (isThreadActive(targetThread)) { yield* waitForBusyThread(application, explicitThreadId); summary = yield* application.getThreadSummary(explicitThreadId); yield* ensureAskTargetAvailable(summary); + targetThread = (yield* application.getThreadMessages({ + threadId: explicitThreadId, + })).thread; } } } From 45ad874824f222acac38e320b7e8dbdfe27c5b02 Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 11:01:34 +0300 Subject: [PATCH 08/11] fix: wait for ask turn completion --- src/cli/ask-lifecycle.ts | 28 ++++++++++++++++++++++++---- 1 file changed, 24 insertions(+), 4 deletions(-) diff --git a/src/cli/ask-lifecycle.ts b/src/cli/ask-lifecycle.ts index d384270..7f190b0 100644 --- a/src/cli/ask-lifecycle.ts +++ b/src/cli/ask-lifecycle.ts @@ -110,6 +110,7 @@ export function waitForAskThread( let lastStatus = ""; let askWindowOpen = false; let answerFound = false; + let turnComplete = false; return Effect.gen(function* () { if (input.format === "human") { yield* output.writeStderr(`waiting for ${input.threadId}...\n`); @@ -118,7 +119,26 @@ export function waitForAskThread( Stream.tap((event) => Effect.gen(function* () { if (event.type === "status") { - yield* ensureNoPendingRequest(application, input.threadId); + const thread = yield* ensureNoPendingRequest(application, input.threadId); + if ( + thread.session?.status === "error" && + input.state.askTurnId !== null && + thread.latestTurn?.turnId === input.state.askTurnId + ) { + yield* Effect.fail( + new ThreadSessionError({ + threadId: input.threadId, + message: thread.session.lastError ?? "thread ended with error", + }), + ); + } + if ( + answerFound && + input.state.askTurnId !== null && + thread.session?.activeTurnId !== input.state.askTurnId + ) { + turnComplete = true; + } } if (event.type === "thread") { const answer = selectAskAnswer(event.thread, input.messageId, input.state.askTurnId); @@ -179,11 +199,11 @@ export function waitForAskThread( } }), ), - Stream.takeUntil((event) => answerFound || event.type === "done"), + Stream.takeUntil((event) => turnComplete || event.type === "done"), Stream.runLast, ); const event = Option.getOrUndefined(last); - if (!answerFound && event?.type !== "done") { + if (!turnComplete && event?.type !== "done") { return yield* Effect.fail( new ThreadSessionError({ message: `thread wait ended without a terminal event: ${input.threadId}`, @@ -312,7 +332,7 @@ function ensureNoPendingRequest(application: T3ApplicationService, threadId: str return application.showThread(threadId).pipe( Effect.flatMap((thread) => { if (!thread.hasPendingApprovals && !thread.hasPendingUserInput) { - return Effect.void; + return Effect.succeed(thread); } return Effect.fail( new AskThreadPendingRequestError({ From 9e57495ac01d8ba1bd81acb7978e7910ad4a249c Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 11:19:50 +0300 Subject: [PATCH 09/11] fix: scope ask turn lifecycle --- src/application/service.ts | 10 ++- src/application/threads.ts | 15 ++--- src/cli/ask-lifecycle.ts | 125 ++++++++++++++++++++++++++----------- src/cli/ask.ts | 11 +++- 4 files changed, 111 insertions(+), 50 deletions(-) diff --git a/src/application/service.ts b/src/application/service.ts index 9792f95..c3101ec 100644 --- a/src/application/service.ts +++ b/src/application/service.ts @@ -86,9 +86,13 @@ export type UpdateThreadInput = { readonly worktreePath?: string | null; }; -export type StartThreadPolicy = { +export interface ThreadDispatchPolicy { readonly until: "dispatch" | "visible" | "complete"; -}; +} + +export interface StartThreadPolicy extends ThreadDispatchPolicy { + readonly onThreadCreated?: (threadId: string) => Effect.Effect<void>; +} export type WaitEvent = | { readonly type: "thread"; readonly thread: OrchestrationThread } @@ -287,7 +291,7 @@ export type T3ThreadApplicationService = { >; readonly sendThread: ( input: SendThreadInput, - policy?: StartThreadPolicy, + policy?: ThreadDispatchPolicy, ) => Effect.Effect< { readonly dispatch: DispatchResult; diff --git a/src/application/threads.ts b/src/application/threads.ts index 0dd14a9..3aeb370 100644 --- a/src/application/threads.ts +++ b/src/application/threads.ts @@ -11,6 +11,8 @@ import { type ListThreadsInclude, type SnoozeThreadInput, type StartThreadInput, + type StartThreadPolicy, + type ThreadDispatchPolicy, } from "./service.ts"; import type { CallbackThreadInput, SendThreadInput } from "./service.ts"; import type { T3ThreadApplicationService } from "./service.ts"; @@ -252,9 +254,7 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function ); const startThread = Effect.fn("T3ApplicationLive.startThread")(function* ( startInput: StartThreadInput, - policy?: { - readonly until: "dispatch" | "visible" | "complete"; - }, + policy?: StartThreadPolicy, ) { const snapshot = yield* orchestration.getShellSnapshot(); const projectRef = startInput.projectRef; @@ -284,10 +284,13 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function project: scope.project, serverConfig, }).pipe(Effect.provideService(Crypto.Crypto, crypto)); + const threadId = commands.threadId; const createDispatch = yield* orchestration.dispatch(commands.createCommand); + if (policy?.onThreadCreated !== undefined) { + yield* policy.onThreadCreated(threadId); + } yield* awaitShellSequence(createDispatch.sequence); const dispatch = yield* orchestration.dispatch(commands.turnCommand); - const threadId = commands.threadId; const messageId = commands.turnCommand.message.messageId; const until = policy?.until ?? "dispatch"; if (until === "dispatch") { @@ -309,9 +312,7 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function }); const sendThread = Effect.fn("T3ApplicationLive.sendThread")(function* ( input: SendThreadInput, - policy?: { - readonly until: "dispatch" | "visible" | "complete"; - }, + policy?: ThreadDispatchPolicy, ) { const modelSelection = input.options !== undefined && input.options.length > 0 diff --git a/src/cli/ask-lifecycle.ts b/src/cli/ask-lifecycle.ts index 7f190b0..d2eb91a 100644 --- a/src/cli/ask-lifecycle.ts +++ b/src/cli/ask-lifecycle.ts @@ -2,7 +2,9 @@ import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; import type { + OrchestrationLatestTurn, OrchestrationMessage, + OrchestrationSession, OrchestrationThread, OrchestrationThreadShell, } from "@t3tools/contracts"; @@ -40,6 +42,7 @@ export type AskArchiveResult = export interface AskExecutionState { readonly archivePolicy: ArchivePolicy; threadId: string | undefined; + createdThread: boolean; dispatched: boolean; askTurnId: string | null; archiveResult: AskArchiveResult | undefined; @@ -79,7 +82,9 @@ export function ensureAskTargetAvailable(thread: OrchestrationThreadShell) { export function waitForBusyThread(application: T3ApplicationService, threadId: string) { return application.watchThread(threadId).pipe( Stream.tap((event) => - event.type === "status" ? ensureNoPendingRequest(application, threadId) : Effect.void, + event.type === "status" + ? application.showThread(threadId).pipe(Effect.flatMap(ensureNoPendingRequest)) + : Effect.void, ), Stream.runLast, Effect.flatMap((last) => { @@ -109,8 +114,50 @@ export function waitForAskThread( ) { let lastStatus = ""; let askWindowOpen = false; - let answerFound = false; + let askMessageCreatedAt: string | null = null; + let askTurnObserved = false; let turnComplete = false; + const observeAskTurnLifecycle = ( + session: OrchestrationSession | null, + latestTurn: OrchestrationLatestTurn | null, + ) => { + if (input.state.askTurnId === null && askWindowOpen) { + const requestedTurnId = + askMessageCreatedAt !== null && latestTurn?.requestedAt === askMessageCreatedAt + ? latestTurn.turnId + : null; + input.state.askTurnId = session?.activeTurnId ?? requestedTurnId; + } + const askTurnId = input.state.askTurnId; + if (askTurnId === null) { + return; + } + const askIsActive = session?.activeTurnId === askTurnId; + const askIsLatest = latestTurn?.turnId === askTurnId; + if (askIsActive || askIsLatest) { + askTurnObserved = true; + } + if (!askTurnObserved) { + return; + } + if (!askIsLatest) { + if ( + askMessageCreatedAt !== null && + latestTurn !== null && + latestTurn.requestedAt > askMessageCreatedAt + ) { + turnComplete = true; + } + return; + } + if (latestTurn.state !== "running") { + turnComplete = true; + return; + } + if (!askIsActive && session?.status !== "starting" && session?.status !== "running") { + turnComplete = true; + } + }; return Effect.gen(function* () { if (input.format === "human") { yield* output.writeStderr(`waiting for ${input.threadId}...\n`); @@ -119,7 +166,8 @@ export function waitForAskThread( Stream.tap((event) => Effect.gen(function* () { if (event.type === "status") { - const thread = yield* ensureNoPendingRequest(application, input.threadId); + const thread = yield* application.showThread(input.threadId); + observeAskTurnLifecycle(thread.session, thread.latestTurn); if ( thread.session?.status === "error" && input.state.askTurnId !== null && @@ -132,26 +180,16 @@ export function waitForAskThread( }), ); } - if ( - answerFound && - input.state.askTurnId !== null && - thread.session?.activeTurnId !== input.state.askTurnId - ) { - turnComplete = true; + if (!turnComplete) { + yield* ensureNoPendingRequest(thread); } } if (event.type === "thread") { - const answer = selectAskAnswer(event.thread, input.messageId, input.state.askTurnId); - if (answer !== undefined) { - if (answer.turnId !== null) { - input.state.askTurnId = answer.turnId; - } - answerFound = true; - } const userIndex = event.thread.messages.findIndex( (message) => message.id === input.messageId && message.role === "user", ); if (userIndex !== -1) { + askMessageCreatedAt = event.thread.messages[userIndex]?.createdAt ?? null; const following = event.thread.messages.slice(userIndex + 1); const nextUserIndex = following.findIndex((message) => message.role === "user"); const askMessages = @@ -161,14 +199,27 @@ export function waitForAskThread( )?.turnId; if (turnId !== undefined && turnId !== null) { input.state.askTurnId = turnId; + askTurnObserved = true; } askWindowOpen = nextUserIndex === -1; + if (!askWindowOpen) { + turnComplete = true; + } + } + const answer = selectAskAnswer(event.thread, input.messageId, input.state.askTurnId); + const answerTurnId = answer?.turnId; + if (answerTurnId !== undefined && answerTurnId !== null) { + input.state.askTurnId = answerTurnId; + askTurnObserved = true; } + observeAskTurnLifecycle(event.thread.session, event.thread.latestTurn); } else if (event.type === "message") { if (event.message.id === input.messageId) { askWindowOpen = true; + askMessageCreatedAt = event.message.createdAt; } else if (askWindowOpen && event.message.role === "user") { askWindowOpen = false; + turnComplete = true; } else if ( askWindowOpen && event.message.role === "assistant" && @@ -179,14 +230,8 @@ export function waitForAskThread( input.state.askTurnId === event.message.turnId ) { input.state.askTurnId = event.message.turnId; - answerFound = !event.message.streaming && event.message.text.trim().length > 0; + askTurnObserved = true; } - } else if ( - askWindowOpen && - event.message.role === "assistant" && - input.state.askTurnId === null - ) { - answerFound = !event.message.streaming && event.message.text.trim().length > 0; } } if (input.format === "ndjson") { @@ -262,7 +307,11 @@ export function finalizeArchive( state.archivePolicy === "always" || (state.archivePolicy === "on-success" && succeeded) || (state.archivePolicy === "on-failure" && !succeeded); - if (!state.dispatched || state.threadId === undefined || !shouldArchive) { + if ( + (!state.createdThread && !state.dispatched) || + state.threadId === undefined || + !shouldArchive + ) { const result = { policy: state.archivePolicy, status: "skipped", @@ -304,12 +353,12 @@ export function cleanupInterruptedAsk( output: T3Output["Service"], state: AskExecutionState, ) { - if (!state.dispatched || state.threadId === undefined) { + if (state.threadId === undefined) { return Effect.void; } const threadId = state.threadId; return Effect.gen(function* () { - if (state.askTurnId !== null) { + if (state.dispatched && state.askTurnId !== null) { yield* application.interruptThreadTurn(threadId, state.askTurnId).pipe( Effect.matchEffect({ onFailure: (error) => @@ -328,18 +377,18 @@ export function ensureTrailingNewline(text: string) { return text.endsWith("\n") ? text : `${text}\n`; } -function ensureNoPendingRequest(application: T3ApplicationService, threadId: string) { - return application.showThread(threadId).pipe( - Effect.flatMap((thread) => { - if (!thread.hasPendingApprovals && !thread.hasPendingUserInput) { - return Effect.succeed(thread); - } - return Effect.fail( - new AskThreadPendingRequestError({ - message: `thread requested approval or user input instead of answering: ${threadId}`, - threadId, - }), - ); +function ensureNoPendingRequest(thread: { + readonly id: string; + readonly hasPendingApprovals: boolean; + readonly hasPendingUserInput: boolean; +}) { + if (!thread.hasPendingApprovals && !thread.hasPendingUserInput) { + return Effect.void; + } + return Effect.fail( + new AskThreadPendingRequestError({ + message: `thread requested approval or user input instead of answering: ${thread.id}`, + threadId: thread.id, }), ); } diff --git a/src/cli/ask.ts b/src/cli/ask.ts index 214954f..0a4e693 100644 --- a/src/cli/ask.ts +++ b/src/cli/ask.ts @@ -174,6 +174,7 @@ export const askCommand = Command.make( const state: AskExecutionState = { archivePolicy, threadId: explicitThreadId, + createdThread: false, dispatched: false, askTurnId: null, archiveResult: undefined, @@ -205,9 +206,15 @@ export const askCommand = Command.make( ...(modelValue !== undefined ? { model: modelValue } : {}), ...(options.length > 0 ? { options } : {}), }, - { until: "dispatch" }, + { + until: "dispatch", + onThreadCreated: (threadId) => + Effect.sync(() => { + state.threadId = threadId; + state.createdThread = true; + }), + }, ); - state.threadId = result.threadId; state.dispatched = true; dispatch = result.dispatch; askMessageId = result.messageId; From 4c1aa2f54c949922400288ba78e9bcbd358045a9 Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 11:54:52 +0300 Subject: [PATCH 10/11] refactor: simplify ask turn tracking --- src/cli/ask-lifecycle.ts | 190 +++++++++++++++++---------------------- src/cli/ask.ts | 20 +++-- 2 files changed, 91 insertions(+), 119 deletions(-) diff --git a/src/cli/ask-lifecycle.ts b/src/cli/ask-lifecycle.ts index d2eb91a..a7a413d 100644 --- a/src/cli/ask-lifecycle.ts +++ b/src/cli/ask-lifecycle.ts @@ -2,9 +2,7 @@ import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; import type { - OrchestrationLatestTurn, OrchestrationMessage, - OrchestrationSession, OrchestrationThread, OrchestrationThreadShell, } from "@t3tools/contracts"; @@ -12,6 +10,7 @@ import type { import type { T3ApplicationService } from "../application/service.ts"; import type { T3CliEnvShape } from "../config/env/env.ts"; import { ThreadSessionError } from "../domain/error.ts"; +import { derivePendingApprovals, derivePendingUserInputs } from "../domain/thread-activities.ts"; import { AskThreadArchivedError, AskThreadPendingRequestError } from "./error.ts"; import { formatWaitEventNdjson } from "./format/thread.ts"; import { isInteractiveHumanTerminal } from "./format/output.ts"; @@ -113,50 +112,20 @@ export function waitForAskThread( }, ) { let lastStatus = ""; - let askWindowOpen = false; - let askMessageCreatedAt: string | null = null; - let askTurnObserved = false; let turnComplete = false; - const observeAskTurnLifecycle = ( - session: OrchestrationSession | null, - latestTurn: OrchestrationLatestTurn | null, - ) => { - if (input.state.askTurnId === null && askWindowOpen) { - const requestedTurnId = - askMessageCreatedAt !== null && latestTurn?.requestedAt === askMessageCreatedAt - ? latestTurn.turnId - : null; - input.state.askTurnId = session?.activeTurnId ?? requestedTurnId; - } - const askTurnId = input.state.askTurnId; - if (askTurnId === null) { - return; - } - const askIsActive = session?.activeTurnId === askTurnId; - const askIsLatest = latestTurn?.turnId === askTurnId; - if (askIsActive || askIsLatest) { - askTurnObserved = true; - } - if (!askTurnObserved) { - return; - } - if (!askIsLatest) { - if ( - askMessageCreatedAt !== null && - latestTurn !== null && - latestTurn.requestedAt > askMessageCreatedAt - ) { - turnComplete = true; - } - return; - } - if (latestTurn.state !== "running") { - turnComplete = true; - return; - } - if (!askIsActive && session?.status !== "starting" && session?.status !== "running") { - turnComplete = true; + const observeAskThread = (thread: OrchestrationThread) => { + const observation = inspectAskTurn(thread, input.messageId, input.state.askTurnId); + input.state.askTurnId = observation.turnId; + if (observation.status === "failed") { + return Effect.fail( + new ThreadSessionError({ + threadId: input.threadId, + message: observation.message, + }), + ); } + turnComplete = observation.status === "complete"; + return Effect.void; }; return Effect.gen(function* () { if (input.format === "human") { @@ -165,73 +134,18 @@ export function waitForAskThread( const last = yield* application.watchThread(input.threadId).pipe( Stream.tap((event) => Effect.gen(function* () { - if (event.type === "status") { - const thread = yield* application.showThread(input.threadId); - observeAskTurnLifecycle(thread.session, thread.latestTurn); - if ( - thread.session?.status === "error" && - input.state.askTurnId !== null && - thread.latestTurn?.turnId === input.state.askTurnId - ) { - yield* Effect.fail( - new ThreadSessionError({ - threadId: input.threadId, - message: thread.session.lastError ?? "thread ended with error", - }), - ); - } + if (event.type === "thread" || event.type === "done") { + yield* observeAskThread(event.thread); + } else if (event.type === "status") { + const thread = (yield* application.getThreadMessages({ threadId: input.threadId })) + .thread; + yield* observeAskThread(thread); if (!turnComplete) { - yield* ensureNoPendingRequest(thread); - } - } - if (event.type === "thread") { - const userIndex = event.thread.messages.findIndex( - (message) => message.id === input.messageId && message.role === "user", - ); - if (userIndex !== -1) { - askMessageCreatedAt = event.thread.messages[userIndex]?.createdAt ?? null; - const following = event.thread.messages.slice(userIndex + 1); - const nextUserIndex = following.findIndex((message) => message.role === "user"); - const askMessages = - nextUserIndex === -1 ? following : following.slice(0, nextUserIndex); - const turnId = askMessages.findLast( - (message) => message.role === "assistant" && message.turnId !== null, - )?.turnId; - if (turnId !== undefined && turnId !== null) { - input.state.askTurnId = turnId; - askTurnObserved = true; - } - askWindowOpen = nextUserIndex === -1; - if (!askWindowOpen) { - turnComplete = true; - } - } - const answer = selectAskAnswer(event.thread, input.messageId, input.state.askTurnId); - const answerTurnId = answer?.turnId; - if (answerTurnId !== undefined && answerTurnId !== null) { - input.state.askTurnId = answerTurnId; - askTurnObserved = true; - } - observeAskTurnLifecycle(event.thread.session, event.thread.latestTurn); - } else if (event.type === "message") { - if (event.message.id === input.messageId) { - askWindowOpen = true; - askMessageCreatedAt = event.message.createdAt; - } else if (askWindowOpen && event.message.role === "user") { - askWindowOpen = false; - turnComplete = true; - } else if ( - askWindowOpen && - event.message.role === "assistant" && - event.message.turnId !== null - ) { - if ( - input.state.askTurnId === null || - input.state.askTurnId === event.message.turnId - ) { - input.state.askTurnId = event.message.turnId; - askTurnObserved = true; - } + yield* ensureNoPendingRequest({ + id: thread.id, + hasPendingApprovals: derivePendingApprovals(thread.activities).length > 0, + hasPendingUserInput: derivePendingUserInputs(thread.activities).length > 0, + }); } } if (input.format === "ndjson") { @@ -294,6 +208,62 @@ export function selectAskAnswer( return candidates.at(-1); } +type AskTurnObservation = + | { readonly status: "waiting"; readonly turnId: string | null } + | { readonly status: "complete"; readonly turnId: string | null } + | { readonly status: "failed"; readonly turnId: string; readonly message: string }; + +export function inspectAskTurn( + thread: OrchestrationThread, + messageId: string, + knownTurnId: string | null, +): AskTurnObservation { + const userIndex = thread.messages.findIndex( + (message) => message.id === messageId && message.role === "user", + ); + const userMessage = userIndex === -1 ? undefined : thread.messages[userIndex]; + if (userMessage === undefined) { + return { status: "waiting", turnId: knownTurnId }; + } + + const following = thread.messages.slice(userIndex + 1); + const nextUserIndex = following.findIndex((message) => message.role === "user"); + const askMessages = nextUserIndex === -1 ? following : following.slice(0, nextUserIndex); + const messageTurnId = + askMessages.findLast((message) => message.role === "assistant" && message.turnId !== null) + ?.turnId ?? null; + const latestTurn = thread.latestTurn; + const requestedTurnId = + latestTurn?.requestedAt === userMessage.createdAt ? latestTurn.turnId : null; + const turnId = + knownTurnId ?? messageTurnId ?? requestedTurnId ?? thread.session?.activeTurnId ?? null; + + if (turnId === null) { + return { + status: nextUserIndex === -1 ? "waiting" : "complete", + turnId, + }; + } + + const askIsLatest = latestTurn?.turnId === turnId; + if (askIsLatest && (latestTurn.state === "error" || thread.session?.status === "error")) { + return { + status: "failed", + turnId, + message: thread.session?.lastError ?? "thread ended with error", + }; + } + + const sessionRunning = + thread.session?.status === "starting" || thread.session?.status === "running"; + const complete = + nextUserIndex !== -1 || + (askIsLatest && + (latestTurn.state !== "running" || + (thread.session?.activeTurnId !== turnId && !sessionRunning))); + return { status: complete ? "complete" : "waiting", turnId }; +} + export function finalizeArchive( application: T3ApplicationService, output: T3Output["Service"], diff --git a/src/cli/ask.ts b/src/cli/ask.ts index 0a4e693..be28c28 100644 --- a/src/cli/ask.ts +++ b/src/cli/ask.ts @@ -16,6 +16,7 @@ import { ensureAskTargetAvailable, ensureTrailingNewline, finalizeArchive, + inspectAskTurn, resolveAskFormat, selectAskAnswer, waitForAskThread, @@ -303,16 +304,17 @@ export const askCommand = Command.make( }); const finalSnapshot = yield* application.getThreadMessages({ threadId }); - const answer = selectAskAnswer(finalSnapshot.thread, askMessageId, state.askTurnId); + const observation = inspectAskTurn(finalSnapshot.thread, askMessageId, state.askTurnId); + if (observation.status === "failed") { + return yield* Effect.fail( + new ThreadSessionError({ + threadId, + message: observation.message, + }), + ); + } + const answer = selectAskAnswer(finalSnapshot.thread, askMessageId, observation.turnId); if (answer === undefined) { - if (finalSnapshot.thread.session?.status === "error") { - return yield* Effect.fail( - new ThreadSessionError({ - threadId, - message: finalSnapshot.thread.session.lastError ?? "thread ended with error", - }), - ); - } return yield* Effect.fail( new AskNoAnswerError({ message: `thread completed without a new final answer: ${threadId}`, From 9fc1125ad88d061b6c88eda78597796ea717d603 Mon Sep 17 00:00:00 2001 From: Tarik02 <Taras.Fomin@gmail.com> Date: Mon, 24 Aug 2026 12:02:49 +0300 Subject: [PATCH 11/11] refactor: reject busy ask threads --- README.md | 3 +- skills/t3code-cli/reference/commands.md | 11 ++--- src/cli/ask-lifecycle.ts | 35 +--------------- src/cli/ask.ts | 54 +++++-------------------- 4 files changed, 17 insertions(+), 86 deletions(-) diff --git a/README.md b/README.md index 3bf6197..38eacd9 100644 --- a/README.md +++ b/README.md @@ -190,8 +190,7 @@ Without `--thread`, it creates a thread and archives it after a successful answe `--thread`, it leaves the existing thread active unless `--archive` is set explicitly. `--archive` accepts `never`, `always`, `on-success`, or `on-failure`. -Existing busy threads fail by default. Use `--busy queue` to wait or `--busy steer` to send the -question immediately. `--timeout 5m` limits the queue and response wait. Unlike other +Existing busy threads are rejected. `--timeout 5m` limits the response wait. Unlike other thread-scoped commands, `ask` uses only an explicit `--thread` and ignores `T3CODE_THREAD_ID`. ### Starting Threads diff --git a/skills/t3code-cli/reference/commands.md b/skills/t3code-cli/reference/commands.md index 3404ec8..f22b772 100644 --- a/skills/t3code-cli/reference/commands.md +++ b/skills/t3code-cli/reference/commands.md @@ -103,7 +103,7 @@ t3cli ask [message] [--title <title>] [--worktree <path>] [--provider <name>] [--model <id>] [--option key=value] [--reasoning-effort <v>] [--effort <v>] [--fast-mode] [--thinking] [--archive never|always|on-success|on-failure] - [--busy fail|queue|steer] [--timeout <duration>] + [--timeout <duration>] [--format auto|human|json|ndjson] t3cli start [message] @@ -128,15 +128,12 @@ t3cli wait [--thread <id>] [--format auto|human|ndjson] archive policy applies to either target. Archive failures produce a warning and remain visible in structured output without changing a successful exit status. -`--busy fail` is the default for an active existing thread. `queue` waits for the active turn, -while `steer` dispatches immediately. Busy checks are best effort because the server does not offer -an atomic busy-policy condition. Existing archived threads and threads with pending approval or -user-input requests are rejected. +Existing busy or archived threads and threads with pending approval or user-input requests are +rejected. `--timeout` accepts positive durations such as `30s`, `5m`, and `1h`; omitting it waits without a limit. A timeout or local interruption stops a turn started by `ask` when ownership can be -confirmed, then applies the failure archive policy. Time spent waiting under `--busy queue` counts -toward the timeout, but a timeout before dispatch does not mutate the existing turn. +confirmed, then applies the failure archive policy. `--title`, `--worktree`, `--provider`, and `--model` apply only when creating a thread and are rejected with `--thread`. Model option flags apply to both modes. `ask` never reads diff --git a/src/cli/ask-lifecycle.ts b/src/cli/ask-lifecycle.ts index a7a413d..b0d9a8e 100644 --- a/src/cli/ask-lifecycle.ts +++ b/src/cli/ask-lifecycle.ts @@ -78,29 +78,6 @@ export function ensureAskTargetAvailable(thread: OrchestrationThreadShell) { return Effect.void; } -export function waitForBusyThread(application: T3ApplicationService, threadId: string) { - return application.watchThread(threadId).pipe( - Stream.tap((event) => - event.type === "status" - ? application.showThread(threadId).pipe(Effect.flatMap(ensureNoPendingRequest)) - : Effect.void, - ), - Stream.runLast, - Effect.flatMap((last) => { - const event = Option.getOrUndefined(last); - if (event?.type === "done") { - return Effect.void; - } - return Effect.fail( - new ThreadSessionError({ - message: `thread wait ended without a terminal event: ${threadId}`, - threadId, - }), - ); - }), - ); -} - export function waitForAskThread( application: T3ApplicationService, output: T3Output["Service"], @@ -174,16 +151,6 @@ export function waitForAskThread( }); } -export function announceQueue(output: T3Output["Service"], format: AskFormat, threadId: string) { - if (format === "ndjson") { - return output.printNdjson({ type: "queue", status: "waiting", threadId }); - } - if (format === "human") { - return output.writeStderr(`thread ${threadId} is busy; waiting to ask...\n`); - } - return Effect.void; -} - export function selectAskAnswer( thread: OrchestrationThread, messageId: string, @@ -236,7 +203,7 @@ export function inspectAskTurn( const requestedTurnId = latestTurn?.requestedAt === userMessage.createdAt ? latestTurn.turnId : null; const turnId = - knownTurnId ?? messageTurnId ?? requestedTurnId ?? thread.session?.activeTurnId ?? null; + knownTurnId ?? requestedTurnId ?? messageTurnId ?? thread.session?.activeTurnId ?? null; if (turnId === null) { return { diff --git a/src/cli/ask.ts b/src/cli/ask.ts index be28c28..a64364a 100644 --- a/src/cli/ask.ts +++ b/src/cli/ask.ts @@ -10,7 +10,6 @@ import { isThreadActive } from "../domain/thread-lifecycle.ts"; import { ThreadSessionError } from "../domain/error.ts"; import { loadT3CliEnv } from "../config/env/env.ts"; import { - announceQueue, archivePolicyChoices, cleanupInterruptedAsk, ensureAskTargetAvailable, @@ -20,7 +19,6 @@ import { resolveAskFormat, selectAskAnswer, waitForAskThread, - waitForBusyThread, } from "./ask-lifecycle.ts"; import type { AskExecutionState } from "./ask-lifecycle.ts"; import { extraArgsConfig } from "./extra-args.ts"; @@ -43,8 +41,6 @@ import { requireCommandProjectRef } from "./require.ts"; import { resolveWorktreePath } from "./scope/index.ts"; import { CliRuntime } from "./runtime/service.ts"; -const busyPolicyChoices = ["fail", "queue", "steer"] as const; - interface AskResponse { readonly answer: string; readonly threadId: string; @@ -65,13 +61,8 @@ const archivePolicyFlag = Flag.choice("archive", archivePolicyChoices).pipe( Flag.optional, ); -const busyPolicyFlag = Flag.choice("busy", busyPolicyChoices).pipe( - Flag.withDescription("Existing-thread busy policy (default: fail)"), - Flag.optional, -); - const timeoutFlag = Flag.string("timeout").pipe( - Flag.withDescription("Queue and response timeout, such as 30s, 5m, or 1h"), + Flag.withDescription("Response timeout, such as 30s, 5m, or 1h"), Flag.optional, ); @@ -93,7 +84,6 @@ export const askCommand = Command.make( model: Flag.string("model").pipe(Flag.optional), ...modelFlags, archive: archivePolicyFlag, - busy: busyPolicyFlag, timeout: timeoutFlag, format: askFormatFlag, ...extraArgsConfig, @@ -114,14 +104,12 @@ export const askCommand = Command.make( fastMode, thinking, archive, - busy, timeout, format, }) => Effect.gen(function* () { const explicitThreadId = Option.getOrUndefined(thread); const created = explicitThreadId === undefined; - const explicitBusyPolicy = Option.getOrUndefined(busy); const titleValue = Option.getOrUndefined(title); const providerValue = Option.getOrUndefined(provider); const modelValue = Option.getOrUndefined(model); @@ -139,13 +127,6 @@ export const askCommand = Command.make( }), ); } - if (created && explicitBusyPolicy !== undefined) { - return yield* Effect.fail( - new InvalidFlagCombinationError({ - message: "--busy can only be used with --thread", - }), - ); - } if (created && force) { return yield* Effect.fail( new InvalidFlagCombinationError({ @@ -158,7 +139,6 @@ export const askCommand = Command.make( const timeoutDuration = timeoutValue === undefined ? undefined : yield* parseAskTimeout(timeoutValue); const archivePolicy = Option.getOrElse(archive, () => (created ? "on-success" : "never")); - const busyPolicy = explicitBusyPolicy ?? "fail"; const application = yield* T3Application; const cliRuntime = yield* CliRuntime; const t3CliEnv = yield* loadT3CliEnv; @@ -220,7 +200,7 @@ export const askCommand = Command.make( dispatch = result.dispatch; askMessageId = result.messageId; } else { - let summary = yield* application.getThreadSummary(explicitThreadId); + const summary = yield* application.getThreadSummary(explicitThreadId); yield* ensureAskTargetAvailable(summary); const projectRef = Option.getOrUndefined(project); @@ -245,28 +225,16 @@ export const askCommand = Command.make( action: "ask", }); - let targetThread = (yield* application.getThreadMessages({ threadId: explicitThreadId })) - .thread; + const targetThread = (yield* application.getThreadMessages({ + threadId: explicitThreadId, + })).thread; if (isThreadActive(targetThread)) { - if (busyPolicy === "fail") { - return yield* Effect.fail( - new AskThreadBusyError({ - message: `thread is busy: ${explicitThreadId}; use --busy queue or --busy steer`, - threadId: explicitThreadId, - }), - ); - } - if (busyPolicy === "queue") { - yield* announceQueue(output, resolvedFormat, explicitThreadId); - while (isThreadActive(targetThread)) { - yield* waitForBusyThread(application, explicitThreadId); - summary = yield* application.getThreadSummary(explicitThreadId); - yield* ensureAskTargetAvailable(summary); - targetThread = (yield* application.getThreadMessages({ - threadId: explicitThreadId, - })).thread; - } - } + return yield* Effect.fail( + new AskThreadBusyError({ + message: `thread is busy: ${explicitThreadId}`, + threadId: explicitThreadId, + }), + ); } const result = yield* application.sendThread(