fix(cli): close batch review findings on billing, path safety, and accounting

- collect: treat a completed batch without request_counts as "results
  expected" instead of failing every item as "no result line" — the old
  guard condemned successful work and invited a paid retry.
- collect: mark an attempt remoteCleaned/collected only when every remote
  deletion succeeded; a failed deletion stays on the retry path (the next
  collect re-fetches the settled batch) instead of leaking files while the
  ledger claims they are gone. The cleanup retry is not re-announced as a
  new settlement.
- retry: gate the plan budget on the real re-assembled requests rather than
  the run-time per-item average, which undercounts retries of large items
  or of sources that grew after run.
- fetch: validate the batch id as a single safe path component before it
  becomes a URL segment and a filename under --out.
- deliver: prove the nearest existing ancestor is inside the project before
  recursive mkdir, so a symlinked parent can no longer create directories
  outside the project; the created parent is still revalidated after.
- deliver the completion content verbatim (trim only guards emptiness):
  leading whitespace and the trailing newline are part of the document.
- refreshTaskStatus: an ambiguous latest attempt outranks partial progress
  so `batch list` says reconcile-first instead of hiding a possibly billing
  submission behind "partial".
- collect --wait: cap each backoff sleep at the remaining --timeout budget
  instead of overshooting the documented deadline.
- batch list: read the local task store without resolving an endpoint or
  requiring credentials, like clean.

packages/cli/src/commands/batch-docs.test.ts

@@ -248,7 +248,18 @@ describe('classifyResult', () => {
       custom_id: 'a#1',
       response: { status_code: 200, body: okBody('# Intro\n') },
     });
-    expect(verdict).toEqual({ kind: 'ok', content: '# Intro' });
+    expect(verdict).toEqual({ kind: 'ok', content: '# Intro\n' });
+  });
+
+  it('delivers the completion verbatim, including meaningful whitespace', () => {
+    // An indented Markdown code block loses its indent under trim(), and the
+    // trailing newline is part of the generated file.
+    const doc = '# Intro\n\n    indented code line\n';
+    const verdict = classifyResult({
+      custom_id: 'a#1',
+      response: { status_code: 200, body: okBody(doc) },
+    });
+    expect(verdict).toEqual({ kind: 'ok', content: doc });
   });
 
   it('rejects a non-200 request status', () => {
@@ -422,6 +433,26 @@ describe('deliverResult', () => {
       fs.rmSync(outside, { recursive: true, force: true });
     }
   });
+
+  it('holds without creating directories outside a symlinked parent', () => {
+    // `linked/new/` does not exist yet: containment must be proven at the
+    // nearest existing ancestor before mkdir creates anything outside.
+    const outside = fs.mkdtempSync(path.join(os.tmpdir(), 'batch-outside-'));
+    try {
+      fs.symlinkSync(outside, path.join(root, 'linked'));
+      const outcome = deliverResult(
+        item({ target: 'linked/new/out.md' }),
+        content,
+        root,
+        sourceHash,
+      );
+      expect(outcome.kind).toBe('held');
+      if (outcome.kind === 'held') expect(outcome.reason).toMatch(/outside/);
+      expect(fs.existsSync(path.join(outside, 'new'))).toBe(false);
+    } finally {
+      fs.rmSync(outside, { recursive: true, force: true });
+    }
+  });
 });
 
 describe('isInsideRoot', () => {

packages/cli/src/commands/batch-docs.ts

@@ -333,7 +333,10 @@ export function classifyResult(line: OutputLine): ResultVerdict {
   if (typeof message.content !== 'string' || !message.content.trim()) {
     return { kind: 'failed', reason: 'empty completion content' };
   }
-  const content = message.content.trim();
+  // Deliver exactly what the model returned: leading whitespace and the
+  // trailing newline can be meaningful (an indented code block), so trim()
+  // only guards emptiness, never edits the document.
+  const content = message.content;
   // A truncated markdown transform most visibly breaks fence pairing; it is
   // a cheap structural signal, not a quality claim.
   if ((content.match(/```/g)?.length ?? 0) % 2 !== 0) {
@@ -392,10 +395,26 @@ export function deliverResult(
     }
   }
   const parent = path.dirname(targetPath);
-  fs.mkdirSync(parent, { recursive: true });
-  // Symlink escape: the parent chain must really live under the project.
-  const realParent = fs.realpathSync(parent);
   const realRoot = fs.realpathSync(projectRoot);
+  // Containment must be proven before anything is created: with a symlink
+  // in the existing chain (linked -> /outside), a recursive mkdir would
+  // otherwise create the missing tail outside the project first.
+  let ancestor = parent;
+  while (!fs.existsSync(ancestor)) {
+    const up = path.dirname(ancestor);
+    if (up === ancestor) break;
+    ancestor = up;
+  }
+  if (!isInsideRoot(realRoot, fs.realpathSync(ancestor))) {
+    return {
+      kind: 'held',
+      reason: `target directory "${item.target}" resolves outside the project root`,
+    };
+  }
+  fs.mkdirSync(parent, { recursive: true });
+  // Revalidate the created parent: the chain must really live under the
+  // project at delivery time, not just before the mkdir.
+  const realParent = fs.realpathSync(parent);
   if (!isInsideRoot(realRoot, realParent)) {
     return {
       kind: 'held',

packages/cli/src/commands/batch-task.test.ts

@@ -318,6 +318,25 @@ describe('refreshTaskStatus', () => {
     expect(task.status).toBe('submit-unknown');
   });
 
+  it('surfaces an ambiguous submission over partial progress', () => {
+    // A retry whose create answer was lost starts from failed/delivered
+    // item states; the list must still say reconcile-first, not "partial".
+    const task = baseTask();
+    task.items[0].state = 'failed';
+    task.attempts.push({
+      attempt: 1,
+      itemIds: ['intro'],
+      submitState: 'created',
+    });
+    task.attempts.push({
+      attempt: 2,
+      itemIds: ['intro'],
+      submitState: 'unknown',
+    });
+    refreshTaskStatus(task);
+    expect(task.status).toBe('submit-unknown');
+  });
+
   it('treats an attempt stuck in uploaded as ambiguous too', () => {
     // A process that died with the create request in flight leaves the
     // attempt `uploaded`; the batch may exist, so it must be reconciled.

packages/cli/src/commands/batch-task.ts

@@ -432,6 +432,14 @@ export function refreshTaskStatus(task: BatchTask): void {
     task.status = 'done';
     return;
   }
+  const last = task.attempts[task.attempts.length - 1];
+  // An ambiguous latest attempt outranks partial progress: its create may
+  // exist and be billing, so the list must say reconcile-first rather than
+  // "some items failed".
+  if (last && isAmbiguous(last)) {
+    task.status = 'submit-unknown';
+    return;
+  }
   if (
     task.attempts.length > 0 &&
     states.some((state) => state !== 'pending' && state !== 'submitted')
@@ -439,10 +447,7 @@ export function refreshTaskStatus(task: BatchTask): void {
     task.status = 'partial';
     return;
   }
-  const last = task.attempts[task.attempts.length - 1];
-  if (last && isAmbiguous(last)) {
-    task.status = 'submit-unknown';
-  } else if (last?.submitState === 'created') {
+  if (last?.submitState === 'created') {
     task.status = 'running';
   }
 }

packages/cli/src/commands/batch-workflow.test.ts

@@ -444,6 +444,26 @@ describe('collectTask', () => {
     expect(task.items.every((item) => item.state === 'delivered')).toBe(true);
   });
 
+  it('waits when a completed batch omits request_counts and has no result file', async () => {
+    const h = (harness = setup());
+    await runPlan(h.deps, h.planPath);
+    const taskId = taskIdOf(h);
+    const job = h.jobs.get('batch-1') as BatchJob;
+    job.status = 'completed';
+    // A provider that omits the counts proves nothing about results:
+    // failing every item here would invite a paid retry of billed work.
+    delete job.request_counts;
+
+    await collectTask(h.deps, taskId);
+    const task = h.store.load(taskId);
+    expect(task.items.map((item) => item.state)).toEqual([
+      'submitted',
+      'submitted',
+    ]);
+    expect(task.attempts[0].collected).toBeUndefined();
+    expect(h.out.join('\n')).toMatch(/result file is not available yet/);
+  });
+
   it('still reports what it delivered when another batch of the task fails to collect', async () => {
     const h = (harness = setup());
     await runAndSettle(h, {
@@ -525,6 +545,39 @@ describe('collectTask', () => {
     ).toBe('# B\n\nBeta.');
   });
 
+  it('retries remote cleanup on the next collect when a deletion fails', async () => {
+    const h = (harness = setup());
+    await runAndSettle(h, {
+      output: `${outputLine('a#1', '# A\n\nAlpha.')}\n${outputLine('b#1', '# B\n\nBeta.')}\n`,
+    });
+    const taskId = taskIdOf(h);
+    h.api.deleteFile.mockRejectedValueOnce(new Error('provider down'));
+    await collectTask(h.deps, taskId);
+    let task = h.store.load(taskId);
+    expect(task.items.every((item) => item.state === 'delivered')).toBe(true);
+    // A failed deletion is not "cleaned", and the attempt is not collected:
+    // marking both anyway would leak the remote files forever.
+    expect(task.attempts[0].remoteCleaned).toBe(false);
+    expect(task.attempts[0].collected).toBe(false);
+    expect(h.err.join('\n')).toMatch(/could not delete remote file file-in-1/);
+
+    const polls = h.api.getBatch.mock.calls.length;
+    const downloads = h.api.downloadFile.mock.calls.length;
+    const summary = await collectTask(h.deps, taskId);
+    task = h.store.load(taskId);
+    expect(task.attempts[0].remoteCleaned).toBe(true);
+    expect(task.attempts[0].collected).toBe(true);
+    // The settled batch was re-fetched to retry cleanup, but results were
+    // not re-downloaded or re-announced.
+    expect(h.api.getBatch.mock.calls.length).toBe(polls + 1);
+    expect(h.api.downloadFile.mock.calls.length).toBe(downloads);
+    expect(summary.settled).toBe(0);
+    expect(summary.delivered).toEqual([]);
+    expect(
+      h.api.deleteFile.mock.calls.map((call: unknown[]) => call[1]),
+    ).toEqual(['file-in-1', 'file-out-1', 'file-in-1', 'file-out-1']);
+  });
+
   it('marks per-item failures from bad statuses, the error file, and missing lines', async () => {
     const h = (harness = setup());
     await runAndSettle(h, {
@@ -778,6 +831,27 @@ describe('collectTask', () => {
     expect(h.sleeps).toEqual([10_000, 20_000]);
     expect(fs.existsSync(path.join(h.root, 'docs', 'en', 'a.md'))).toBe(true);
   });
+
+  it('caps each backoff sleep at the remaining --timeout budget', async () => {
+    const h = (harness = setup());
+    await runPlan(h.deps, h.planPath);
+    let fakeNow = Date.now();
+    const nowSpy = vi.spyOn(Date, 'now').mockImplementation(() => fakeNow);
+    h.deps.sleep = async (ms) => {
+      h.sleeps.push(ms);
+      fakeNow += ms;
+    };
+    try {
+      await expect(
+        collectTask(h.deps, taskIdOf(h), { wait: true, timeoutSeconds: 15 }),
+      ).rejects.toThrow(/still in_progress after the --wait timeout/);
+    } finally {
+      nowSpy.mockRestore();
+    }
+    // 10s then 20s of backoff, but only 15s of budget: the second sleep is
+    // capped at the remaining 5s instead of overshooting the deadline.
+    expect(h.sleeps).toEqual([10_000, 5_000]);
+  });
 });
 
 describe('retryTask', () => {
@@ -852,6 +926,29 @@ describe('retryTask', () => {
     expect(h.api.uploadJsonl).toHaveBeenCalledTimes(1);
   });
 
+  it('gates the retry on the real assembly, not the run-time average', async () => {
+    const h = (harness = setup({ maxCostUsd: 0.01 }));
+    h.deps.env = {
+      ...h.deps.env,
+      QWEN_BATCH_INPUT_PRICE_PER_1M_USD: '2',
+      QWEN_BATCH_OUTPUT_PRICE_PER_1M_USD: '6',
+    };
+    await runAndSettle(h, {
+      output: `${outputLine('a#1', '# A\n\nAlpha.')}\n${JSON.stringify({ custom_id: 'b#1', response: { status_code: 500, body: {} } })}\n`,
+    });
+    const taskId = taskIdOf(h);
+    await collectTask(h.deps, taskId);
+    // b's source grows massively after the run: the stale per-item average
+    // in task.estimate stays tiny, but the request actually being submitted
+    // is what the budget must judge.
+    fs.writeFileSync(
+      path.join(h.root, 'docs', 'zh', 'b.md'),
+      'y'.repeat(30000),
+    );
+    await expect(retryTask(h.deps, taskId)).rejects.toThrow(/exceeds the plan/);
+    expect(h.api.uploadJsonl).toHaveBeenCalledTimes(1);
+  });
+
   it('skips truncated items unless a larger output limit is given', async () => {
     const h = (harness = setup({ maxOutputTokens: 1000 }));
     await runAndSettle(h, {

packages/cli/src/commands/batch-workflow.ts

@@ -677,7 +677,9 @@ async function waitForSettled(
       );
     }
     deps.out(`waiting: ${batchId} is ${job.status} …`);
-    await sleep(delay);
+    // Cap the nap at the remaining deadline: --timeout is a hard ceiling on
+    // the whole collect, not a per-iteration suggestion.
+    await sleep(Math.min(delay, deadline - Date.now()));
     delay = Math.min(delay * 2, 60_000);
   }
 }
@@ -754,6 +756,10 @@ async function collectLocked(
     // deliveries without asking the provider about a batch it may have
     // forgotten by now.
     let job: BatchJob | undefined;
+    // finalStatus is recorded at the first harvest; a later collect that
+    // only retries remote cleanup must not re-announce the batch as newly
+    // settled (auto-collect would repeat the notice every pass).
+    const firstHarvest = attempt.finalStatus === undefined;
     if (!attempt.collected) {
       job = await api.getBatch(deps.ep, batchId);
       if (!SETTLED_STATUSES.has(job.status)) {
@@ -769,12 +775,14 @@ async function collectLocked(
     }
 
     // Settled but its result file not published yet: collecting now would
-    // fail every item as "no result line" and invite a paid retry.
+    // fail every item as "no result line" and invite a paid retry. A
+    // provider that omits request_counts is treated as "results expected"
+    // (fail closed): defer the collection rather than condemn the items.
     if (
       job?.status === 'completed' &&
       !job.output_file_id &&
       !job.error_file_id &&
-      (job.request_counts?.completed ?? 0) > 0
+      job.request_counts?.completed !== 0
     ) {
       deps.out(
         `${batchId} is completed but its result file is not available yet; collect again shortly.`,
@@ -930,6 +938,7 @@ async function collectLocked(
       // Results are safely local now; uploaded files otherwise live on the
       // provider until somebody deletes them. Failure here must not fail
       // the collection — the report below stays truthful either way.
+      let failed = 0;
       for (const fileId of [
         job.input_file_id,
         job.output_file_id,
@@ -939,16 +948,22 @@ async function collectLocked(
         try {
           await api.deleteFile(deps.ep, fileId);
         } catch (error) {
+          failed += 1;
           deps.err(
             `[batch] warning: could not delete remote file ${fileId}: ${error instanceof Error ? error.message : String(error)}`,
           );
         }
       }
-      attempt.remoteCleaned = true;
+      // Cleanup counts only when every deletion landed; otherwise the
+      // attempt stays uncollected below, so the next collect re-fetches the
+      // settled batch and retries instead of leaking the files forever.
+      attempt.remoteCleaned = failed === 0;
     }
     if (job) {
-      attempt.collected = true;
-      settledNow += 1;
+      // `collect --keep-remote` opts out of cleanup entirely.
+      if (firstHarvest) settledNow += 1;
+      attempt.collected =
+        options.keepRemote === true || attempt.remoteCleaned === true;
       store.save(task);
     }
   };
@@ -1139,24 +1154,24 @@ async function retryLocked(
     ...(newLimit === undefined ? {} : { maxOutputTokens: newLimit }),
   };
   task.attempts.push(attempt);
-  const perItem = task.estimate
-    ? {
-        input: task.estimate.inputTokens / task.items.length,
-        output: task.estimate.outputTokens / task.items.length,
-      }
-    : { input: 0, output: 0 };
-  const cost = costLine(
-    Math.round(perItem.input * retryItems.length),
-    Math.round(perItem.output * retryItems.length),
-    deps.env,
+  // The budget must bind what is actually submitted: assemble the real
+  // retry requests (sources are re-read now) instead of extrapolating from
+  // the original run's per-item average, which undercounts a retry whose
+  // failed items are the large ones — or whose sources grew since `run`.
+  const assembly = assembleAttempt(
+    task,
+    attemptNumber,
+    attempt.itemIds,
+    attempt.maxOutputTokens,
   );
+  const cost = costLine(assembly.inputTokens, assembly.outputTokens, deps.env);
   // The plan's budget binds every submission, not just the first.
   enforceBudget(task.plan, cost);
   deps.out(
     `retrying ${retryItems.length} item(s) as attempt ${attemptNumber}` +
       `${newLimit === undefined ? '' : ` with max output ${newLimit} tokens`}: ${cost.text}`,
   );
-  const job = await submitAttempt(deps, task, attempt, store);
+  const job = await submitAttempt(deps, task, attempt, store, assembly);
   await reportEarlyRejection(deps, task.id, job);
   deps.out(`collect later with: qwen batch collect ${task.id}`);
 }
@@ -1189,7 +1204,10 @@ export async function checkReadiness(deps: WorkflowDeps): Promise<void> {
   );
 }
 
-export async function listTasks(deps: WorkflowDeps): Promise<void> {
+/** Local only: reads the task store; no endpoint or credentials needed. */
+export async function listTasks(
+  deps: Pick<WorkflowDeps, 'env' | 'out'>,
+): Promise<void> {
   const store = new BatchTaskStore(batchHomeDir(deps.env));
   const tasks = store.list();
   if (tasks.length === 0) {

packages/cli/src/commands/batch.test.ts

@@ -600,6 +600,18 @@ describe('submitBatch / fetchBatch', () => {
     expect(fs.existsSync(path.join(dir, 'b.output.jsonl.part'))).toBe(false);
   });
 
+  it('refuses a batch id that is not a safe filename component', async () => {
+    // The id becomes `<id>.output.jsonl` under outDir and a URL segment:
+    // separators or dots-only prefixes would write outside the directory.
+    await expect(fetchBatch(ep, '../escape', dir, false)).rejects.toThrow(
+      /invalid batch id/,
+    );
+    await expect(fetchBatch(ep, 'a/b', dir, false)).rejects.toThrow(
+      /invalid batch id/,
+    );
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
   it('still reports a successful fetch when a remote delete fails', async () => {
     // The results are already on disk; a failed DELETE must not abort the
     // command before the written paths reach the user.

packages/cli/src/commands/batch.ts

@@ -7,8 +7,8 @@
 // File for 'qwen batch' — submit, inspect, fetch, and cancel DashScope Batch
 // API jobs. Batch runs at half the realtime price with a >=24h completion
 // window, so it is a fan-out tool for many independent single-turn requests,
-// not a path for the agent loop. Rationale and probe results:
-// docs/plans/2026-09-14-batch-api-feasibility.md
+// not a path for the agent loop. Rationale and measurements:
+// docs/design/2026-09-23-agent-prepared-batch-api.md
 import fs from 'node:fs';
 import path from 'node:path';
 import readline from 'node:readline';
@@ -441,6 +441,14 @@ export async function fetchBatch(
   outDir: string,
   remove: boolean,
 ): Promise<{ job: BatchJob; written: string[] }> {
+  // The argv id becomes a URL segment and a filename verbatim: a value with
+  // path separators (../../report) would write outside outDir, and the
+  // download's final rename would replace whatever lives there.
+  if (!/^[A-Za-z0-9][A-Za-z0-9._-]*$/.test(id)) {
+    throw new Error(
+      `invalid batch id "${id}": expected a single id component (letters, digits, dot, dash, underscore)`,
+    );
+  }
   const job = await getBatch(ep, id);
   if (!SETTLED.has(job.status)) {
     throw new Error(
@@ -773,11 +781,10 @@ const listWorkflowCommand: CommandModule = {
   command: 'list',
   describe: 'List all recorded workflow tasks with their project',
   builder: (yargs) => yargs,
-  handler: (argv) =>
+  // Local only: no endpoint or credentials needed.
+  handler: () =>
     run(async () => {
-      await listTasks(
-        workflowDeps(await prepareEndpoint(process.env, cliOptionsOf(argv))),
-      );
+      await listTasks({ env: process.env, out: writeStdoutLine });
     }),
 };