Add td returned: list and watch your functions that came back, with why

src/client/cli.js

@@ -39,6 +39,8 @@ Work (a region: a source file or class, done as a whole)
   td task <taskId>                            show a task's context
   td submit [taskId...] [--files a.cpp,b.h]   submit everything you hold (or these tasks, or only these files)
   td batch <batchId>                          state of a submitted batch, per function
+  td returned [--watch] [--json]              your functions that came back (sent back, rejected, returned), per batch,
+                                              with why; --watch polls and prints new ones as they come back
   td result <attemptId>                       show a verification result
   td withdraw <attemptId>                     take back a function waiting for review, to fix it
   td heartbeat [taskId] | td release <taskId> extend your leases (all without an id)
@@ -79,6 +81,36 @@ function printCheck(r) {
   for (const c of r.referenceConflicts) console.log(`conflict: ${c.message ?? c}`);
 }
 
+function printReturned(r) {
+  for (const b of r.batches) {
+    console.log(`\nbatch ${b.batchId}${b.returned ? `: ${b.returned}` : ''}`);
+    for (const f of b.functions) {
+      const score = f.score == null ? '-' : `${(f.score * 100).toFixed(1)}%`;
+      const held = f.taskState === 'leased' ? ` held by ${f.holder}` : ` task ${f.taskState}`;
+      console.log(`  #${f.taskId} ${f.symbol} ${(f.name ?? '').padEnd(30)} ${f.state.padEnd(13)} ${score.padStart(6)} attempt ${f.attemptId}${held}  ${f.unit ?? ''}`);
+      if (f.summary) console.log(`      ${f.summary}`);
+      if (f.peerOutcome) console.log(`      ${f.peerOutcome}`);
+      for (const v of f.peerDifferent) console.log(`      different (${v.reviewer}): ${v.reason?.text ?? v.reason ?? ''}`);
+      for (const c of f.reviewerComments) console.log(`      ${c.action}: ${c.comment?.text ?? ''}`);
+    }
+  }
+  console.log(`\n${r.functions} function(s) to fix in ${r.batches.length} batch(es); td claim <taskId>, fix, td check, td submit <taskIds> --files ...`);
+}
+
+async function watchReturned(client, json) {
+  const seen = new Set();
+  for (;;) {
+    const r = await client.get('/api/me/returned');
+    const fresh = r.batches.map((b) => ({ ...b, functions: b.functions.filter((f) => !seen.has(f.attemptId)) })).filter((b) => b.functions.length);
+    for (const b of r.batches) for (const f of b.functions) seen.add(f.attemptId);
+    if (fresh.length) {
+      const out = { functions: fresh.reduce((n, b) => n + b.functions.length, 0), batches: fresh };
+      json ? console.log(JSON.stringify(out)) : (console.log(`== ${new Date().toISOString()}`), printReturned(out));
+    }
+    await new Promise((done) => setTimeout(done, 60_000));
+  }
+}
+
 /** How an MCP client starts us: the td binary itself, or node + this script when run from source. */
 async function launcher() {
   const sea = await import('node:sea').catch(() => null);
@@ -133,6 +165,7 @@ async function main() {
       force: { type: 'boolean', default: false },
       workspace: { type: 'string' },
       help: { type: 'boolean', short: 'h' },
+      watch: { type: 'boolean', default: false },
     },
   });
   const [cmd, ...args] = positionals;
@@ -181,6 +214,9 @@ connect your AI with:  td connect --workspace "${ws.dir}"`);
       return print(args[0] ? await client.post(`/api/tasks/${args[0]}/heartbeat`) : await client.post('/api/tasks/heartbeat-all'));
     case 'batch':
       return print(await client.get(`/api/batches/${args[0]}`));
+    case 'returned':
+      if (values.watch) return watchReturned(client, values.json);
+      return values.json ? print(await client.get('/api/me/returned')) : printReturned(await client.get('/api/me/returned'));
     case 'release':
       return print(await client.post(`/api/tasks/${args[0]}/release`));
     case 'withdraw':

src/client/mcp.js

@@ -208,6 +208,12 @@ const TOOLS = [
     inputSchema: { type: 'object', properties: { batchId: int('batch id') }, required: ['batchId'] },
     run: async (ws, a) => ops.api(ws).get(`/api/batches/${a.batchId}`),
   },
+  {
+    name: 'td_returned',
+    description: 'Your functions that came back and still need work (sent back by peers or reviewers, rejected at a gate, returned with their batch), grouped by batch: task id, attempt id, why (summary, peer "different" reasons, reviewer comments) and whether the task is held. Fix them with td_claim_task, td_check and td_submit; read details with td_get_result.',
+    inputSchema: { type: 'object', properties: { since: int('only attempts finished at or after this time (ms since epoch)') } },
+    run: async (ws, a) => ops.api(ws).get(`/api/me/returned${a.since ? `?since=${a.since}` : ''}`),
+  },
   {
     name: 'td_get_result',
     description: 'Get the verification result of a submitted attempt (state queued/verifying until done).',

src/server/app.js

@@ -8,7 +8,7 @@ import { fileURLToPath } from 'node:url';
 import { Store, now } from './db.js';
 import { Accounts, AccountError, hasRole, ROLES } from './accounts.js';
 import { Repo, serveGit } from './git.js';
-import { Tasks } from './tasks.js';
+import { Tasks, untrusted } from './tasks.js';
 import { Verifier } from './verify.js';
 import { Merger } from './merge.js';
 import { Maintenance } from './maintenance.js';
@@ -638,6 +638,53 @@ export function createApp(ctx) {
     });
   });
 
+  // The caller's functions that came back and still need work: the latest attempt of each task,
+  // sent back, gate-rejected or returned with its batch, while the task is not done. Grouped by batch.
+  router.on('GET', '/api/me/returned', ({ req, query }) => {
+    const user = auth(req);
+    const since = query.get('since') ? intParam(query.get('since')) : 0;
+    const limit = Math.min(500, Math.max(1, intParam(query.get('limit') ?? '200')));
+    const rows = store.all(
+      `SELECT a.id, a.batch_id, a.task_id, a.state, a.score, a.finished_at, a.result_json, t.symbol_id, t.state AS task_state, t.cleanup_cycles,
+              s.name, s.unit, lu.name AS holder, b.result_json AS batch_result
+         FROM attempts a JOIN tasks t ON t.id = a.task_id JOIN symbols s ON s.id = t.symbol_id
+         LEFT JOIN users lu ON lu.id = t.lease_user AND t.state = 'leased'
+         LEFT JOIN batches b ON b.id = a.batch_id
+        WHERE a.user_id = ? AND a.state IN ('needs_cleanup', 'rejected', 'returned') AND COALESCE(a.finished_at, a.created_at) >= ?
+          AND t.state IN ('open', 'leased', 'needs_cleanup', 'hard')
+          AND NOT EXISTS (SELECT 1 FROM attempts n WHERE n.task_id = a.task_id AND n.id > a.id)
+        ORDER BY a.batch_id DESC, a.id LIMIT ?`,
+      user.id, since, limit,
+    );
+    const batches = new Map();
+    for (const r of rows) {
+      if (!batches.has(r.batch_id)) {
+        const br = r.batch_result ? JSON.parse(r.batch_result) : {};
+        batches.set(r.batch_id, { batchId: r.batch_id, returned: br.returned ?? null, functions: [] });
+      }
+      const res = r.result_json ? JSON.parse(r.result_json) : {};
+      const fb = res.feedback ?? {};
+      batches.get(r.batch_id).functions.push({
+        attemptId: r.id,
+        taskId: r.task_id,
+        symbol: r.symbol_id,
+        name: r.name,
+        unit: r.unit,
+        state: r.state,
+        score: r.score,
+        taskState: r.task_state,
+        holder: r.holder ?? null,
+        cleanupCycles: r.cleanup_cycles,
+        finishedAt: r.finished_at,
+        summary: fb.summary ?? res.returned ?? null,
+        peerOutcome: res.peerOutcome ?? null,
+        peerDifferent: peers.verdicts(r.id).filter((v) => v.verdict === 'different').map((v) => ({ reviewer: v.reviewer, reason: v.reason })),
+        reviewerComments: store.all(`SELECT r.action, r.comment FROM reviews r WHERE r.attempt_id = ? AND r.action != 'approve'`, r.id).map((c) => ({ action: c.action, comment: untrusted(c.comment) })),
+      });
+    }
+    return guided({ functions: rows.length, batches: [...batches.values()] });
+  });
+
   router.on('POST', '/api/tasks/heartbeat-all', ({ req }) => {
     const user = auth(req);
     return { ...tasks.heartbeatAll(user), regions: ctx.regions.heartbeat(user) };

test/server.test.js

@@ -1938,6 +1938,47 @@ test('the batch list shows every batch, also those sent back before review, filt
   ctx.store.run(`UPDATE attempts SET state = 'rejected' WHERE batch_id = ?`, sentBack);
 });
 
+test('returned lists the latest failed attempt of each unfinished task of the caller, per batch, with the reasons', async () => {
+  const token = await activeUser('rlister');
+  const me = ctx.accounts.byName('rlister');
+  await activeUser('rlother');
+  const other = ctx.accounts.byName('rlother');
+  const head = ctx.repo.head();
+  const task = (symbol) => {
+    ctx.store.run(`INSERT INTO symbols (id, kind, name, created_at) VALUES (?, 'function', ?, ?)`, symbol, symbol, now());
+    return { id: Number(ctx.store.run(`INSERT INTO tasks (symbol_id, state, updated_at) VALUES (?, 'open', ?)`, symbol, now()).lastInsertRowid), state: 'done' };
+  };
+  const [fixMe, fixedLater, done] = ['fn_rl_1', 'fn_rl_2', 'fn_rl_3'].map(task);
+  const batch = (user, result) => Number(ctx.store.run(`INSERT INTO batches (user_id, base_commit, files_json, state, result_json, created_at) VALUES (?, ?, '{}', 'returned', ?, ?)`, user.id, head, JSON.stringify(result), now()).lastInsertRowid);
+  const attempt = (user, task, state, batchId, feedback) =>
+    Number(ctx.store.run(`INSERT INTO attempts (task_id, user_id, base_commit, files_json, state, batch_id, result_json, created_at, finished_at) VALUES (?, ?, ?, '{}', ?, ?, ?, ?, ?)`, task.id, user.id, head, state, batchId, JSON.stringify({ feedback }), now(), now()).lastInsertRowid);
+  try {
+    const b = batch(me, { returned: '2 of 3 functions need work' });
+    const a = attempt(me, fixMe, 'needs_cleanup', b, { summary: 'peers sent it back' });
+    attempt(me, fixedLater, 'needs_cleanup', b, { summary: 'compilation failed' });
+    attempt(me, done, 'rejected', b, { summary: 'rejected by the static gate' });
+    attempt(other, fixedLater, 'peer_review', batch(other, {}), { summary: 'not byte-identical' });
+    ctx.store.run(`UPDATE tasks SET state = 'needs_cleanup' WHERE id = ?`, fixMe.id);
+    ctx.store.run(`UPDATE tasks SET state = 'done' WHERE id = ?`, done.id);
+    ctx.store.run(`INSERT INTO peer_reviews (attempt_id, reviewer_id, verdict, reason, created_at) VALUES (?, ?, 'different', 'the original checks the flag first', ?)`, a, other.id, now());
+
+    assert.equal((await call('GET', '/api/me/returned')).status, 401);
+    const r = (await call('GET', '/api/me/returned', null, token)).body;
+    assert.equal(r.functions, 1, 'a newer attempt and a done task are not work to fix');
+    assert.equal(r.batches.length, 1);
+    assert.equal(r.batches[0].batchId, b);
+    assert.equal(r.batches[0].returned, '2 of 3 functions need work');
+    const [f] = r.batches[0].functions;
+    assert.deepEqual([f.attemptId, f.taskId, f.state, f.taskState, f.summary], [a, fixMe.id, 'needs_cleanup', 'needs_cleanup', 'peers sent it back']);
+    assert.equal(f.peerDifferent[0].reason.untrusted, true);
+    assert.equal(f.peerDifferent[0].reason.text, 'the original checks the flag first');
+    assert.equal((await call('GET', `/api/me/returned?since=${now() + 60_000}`, null, token)).body.functions, 0);
+    assert.equal((await call('GET', '/api/me/returned', null, await activeUser('rlister2'))).body.functions, 0, 'only the caller\'s own attempts');
+  } finally {
+    for (const t of [fixMe, fixedLater, done]) ctx.store.run(`UPDATE tasks SET state = ? WHERE id = ?`, t.state, t.id);
+  }
+});
+
 test('review batch filtering precedes pagination and ignores stale open envelopes', async () => {
   const token = await activeUser('batchfilter');
   const user = ctx.accounts.byName('batchfilter');