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..38eacd9 100644 --- a/README.md +++ b/README.md @@ -177,6 +177,22 @@ 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 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 ```sh @@ -304,6 +320,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/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/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..f22b772 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] + [--timeout <duration>] + [--format auto|human|json|ndjson] + t3cli start [message] [--project <ref>] [--stdin] [--title <title>] [--worktree <path>] [--provider <name>] [--model <id>] @@ -115,6 +123,22 @@ 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. + +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. + +`--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 +209,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 +227,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..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 } @@ -209,6 +213,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>; @@ -227,6 +232,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; @@ -246,6 +254,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: ( @@ -270,6 +282,7 @@ export type T3ThreadApplicationService = { ) => Effect.Effect< { readonly dispatch: DispatchResult; + readonly messageId: string; readonly project: OrchestrationProjectShell; readonly threadId: string; readonly thread?: OrchestrationThread; @@ -278,10 +291,11 @@ export type T3ThreadApplicationService = { >; readonly sendThread: ( input: SendThreadInput, - policy?: StartThreadPolicy, + policy?: ThreadDispatchPolicy, ) => 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 a182fd8..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"; @@ -117,6 +119,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 +144,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, @@ -188,6 +209,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), @@ -220,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; @@ -252,13 +284,17 @@ 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") { - return { dispatch, project: scope.project, threadId }; + return { dispatch, messageId, project: scope.project, threadId }; } yield* awaitShellSequence(dispatch.sequence); if (until === "visible") { @@ -268,17 +304,15 @@ 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, - policy?: { - readonly until: "dispatch" | "visible" | "complete"; - }, + policy?: ThreadDispatchPolicy, ) { const modelSelection = input.options !== undefined && input.options.length > 0 @@ -292,9 +326,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") { @@ -304,11 +339,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) { @@ -352,8 +387,10 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function return { approveThread, archiveThread, + awaitShellSequence, deleteThread, interruptThread, + interruptThreadTurn, pinThread, settleThread, snoozeThread, @@ -365,6 +402,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..b0d9a8e --- /dev/null +++ b/src/cli/ask-lifecycle.ts @@ -0,0 +1,331 @@ +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 { 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"; +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 archivePolicy: ArchivePolicy; + threadId: string | undefined; + createdThread: boolean; + dispatched: boolean; + askTurnId: 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 waitForAskThread( + application: T3ApplicationService, + output: T3Output["Service"], + input: { + readonly threadId: string; + readonly format: AskFormat; + readonly messageId: string; + readonly state: AskExecutionState; + }, +) { + let lastStatus = ""; + let turnComplete = false; + 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") { + 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 === "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({ + id: thread.id, + hasPendingApprovals: derivePendingApprovals(thread.activities).length > 0, + hasPendingUserInput: derivePendingUserInputs(thread.activities).length > 0, + }); + } + } + 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.takeUntil((event) => turnComplete || event.type === "done"), + Stream.runLast, + ); + const event = Option.getOrUndefined(last); + if (!turnComplete && event?.type !== "done") { + return yield* Effect.fail( + new ThreadSessionError({ + message: `thread wait ended without a terminal event: ${input.threadId}`, + threadId: input.threadId, + }), + ); + } + return yield* Effect.void; + }); +} + +export function selectAskAnswer( + thread: OrchestrationThread, + messageId: string, + turnId: string | null, +): OrchestrationMessage | undefined { + const userIndex = thread.messages.findIndex( + (message) => message.id === messageId && message.role === "user", + ); + if (userIndex === -1) { + return undefined; + } + 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); +} + +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 ?? requestedTurnId ?? messageTurnId ?? 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"], + 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.createdThread && !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.threadId === undefined) { + return Effect.void; + } + const threadId = state.threadId; + return Effect.gen(function* () { + if (state.dispatched && state.askTurnId !== null) { + yield* application.interruptThreadTurn(threadId, state.askTurnId).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(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 new file mode 100644 index 0000000..a64364a --- /dev/null +++ b/src/cli/ask.ts @@ -0,0 +1,390 @@ +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 { + archivePolicyChoices, + cleanupInterruptedAsk, + ensureAskTargetAvailable, + ensureTrailingNewline, + finalizeArchive, + inspectAskTurn, + resolveAskFormat, + selectAskAnswer, + waitForAskThread, +} 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"; + +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 timeoutFlag = Flag.string("timeout").pipe( + Flag.withDescription("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, + timeout: timeoutFlag, + format: askFormatFlag, + ...extraArgsConfig, + }, + ({ + project, + thread, + force, + message, + stdin, + title, + worktree, + provider, + model, + option, + reasoningEffort, + effort, + fastMode, + thinking, + archive, + timeout, + format, + }) => + Effect.gen(function* () { + const explicitThreadId = Option.getOrUndefined(thread); + const created = explicitThreadId === undefined; + 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 && 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 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 = { + archivePolicy, + threadId: explicitThreadId, + createdThread: false, + dispatched: false, + askTurnId: null, + archiveResult: undefined, + }; + + const response = Effect.gen(function* () { + const text = yield* readInitialMessage({ + message: Option.getOrUndefined(message), + fromStdin: stdin, + readStdin: input.readStdin, + }); + let dispatch: DispatchResult; + let askMessageId: string; + + 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", + onThreadCreated: (threadId) => + Effect.sync(() => { + state.threadId = threadId; + state.createdThread = true; + }), + }, + ); + state.dispatched = true; + dispatch = result.dispatch; + askMessageId = result.messageId; + } else { + const 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", + }); + + const targetThread = (yield* application.getThreadMessages({ + threadId: explicitThreadId, + })).thread; + if (isThreadActive(targetThread)) { + return yield* Effect.fail( + new AskThreadBusyError({ + message: `thread is busy: ${explicitThreadId}`, + threadId: explicitThreadId, + }), + ); + } + + const result = yield* application.sendThread( + { + threadId: explicitThreadId, + message: text, + ...(options.length > 0 ? { options } : {}), + }, + { until: "dispatch" }, + ); + state.dispatched = true; + dispatch = result.dispatch; + askMessageId = result.messageId; + } + + 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* application.awaitShellSequence(dispatch.sequence); + yield* waitForAskThread(application, output, { + threadId, + format: resolvedFormat, + messageId: askMessageId, + state, + }); + + const finalSnapshot = yield* application.getThreadMessages({ threadId }); + 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) { + 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", { 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({