From 1038c1d390d6376b769ea5ac8a69151e723cd031 Mon Sep 17 00:00:00 2001 From: Saty <5752427+SatyPardus@users.noreply.github.com> Date: Tue, 20 Oct 2026 03:21:56 +0200 Subject: [PATCH] Retire single-function and server-picked claims; merge rate in bytes per hour - New work is claimed by region only: claim without a task claims the best free region, claim-batch with a file claims it as a region, address ranges are refused; a single function only to fix one that came back. td_claim_batch and td_list_tasks are gone from the MCP tools. - Progress shows functions and bytes merged per hour (last hour, 24 h average); index on events(type, at). Co-Authored-By: Claude Opus 5.5 --- src/client/cli.js | 12 +++------ src/client/mcp.js | 16 ++---------- src/client/ops.js | 8 ++---- src/server/app.js | 55 +++++++++++++++++++++++++--------------- src/server/db.js | 1 + src/server/web/app.js | 14 ++++++++++ src/server/web/style.css | 7 +++++ test/server.test.js | 32 +++++++++++++++++------ 8 files changed, 88 insertions(+), 57 deletions(-) diff --git a/src/client/cli.js b/src/client/cli.js index 27ac64c..850b06f 100644 --- a/src/client/cli.js +++ b/src/client/cli.js @@ -35,9 +35,7 @@ Work (a region: a source file or class, done as a whole) unit that includes a changed header, as the merge does) td submit [--notes T] submit the region's work as one batch td tasks [--state S] [--limit N] list claimable tasks - td claim [taskId] claim a task, prints its context - td claim --unit [--limit N] claim every open function of a source file (a batch) - td claim --range - [--limit N] claim every open function of an address range + td claim claim a function that came back, to fix it td task show a task's context td submit [taskId...] [--notes T] submit everything you hold (or these tasks) as one batch td batch state of a submitted batch, per function @@ -161,11 +159,9 @@ connect your AI with: td connect --workspace "${ws.dir}"`); return; case 'claim': { // td claim [taskId] | td claim --unit | td claim --range 0x140001000-0x140002000 [--limit N] - if (values.unit || values.range) { - const [start, end] = values.range ? values.range.split('-') : []; - return print(await ops.claimBatch(ws, { unit: values.unit ? ops.relPath(ws, values.unit) : undefined, start, end, limit: Number(values.limit) || 50 })); - } - return print(await client.post('/api/tasks/claim', { taskId: args[0] ? Number(args[0]) : undefined })); + if (values.unit) return print(await ops.claimRegion(ws, { kind: 'unit', key: ops.relPath(ws, values.unit) })); + if (!args[0]) throw new Error('claim new work by region: td regions, then td region claim ; td claim fixes a function that came back'); + return print(await client.post('/api/tasks/claim', { taskId: Number(args[0]) })); } case 'task': return print(await client.get(`/api/tasks/${args[0]}`)); diff --git a/src/client/mcp.js b/src/client/mcp.js index af798a7..f326c89 100644 --- a/src/client/mcp.js +++ b/src/client/mcp.js @@ -38,16 +38,10 @@ const TOOLS = [ inputSchema: { type: 'object', properties: {} }, run: async (ws) => ops.toolchainInstall(ws, { log: () => {} }), }, - { - name: 'td_list_tasks', - description: 'List claimable decompilation tasks (easiest first).', - inputSchema: { type: 'object', properties: { state: str('comma-separated states, default open,needs_cleanup,hard'), unit: str('only tasks in this translation unit'), limit: int('max results, default 20') } }, - run: async (ws, a) => ops.api(ws).get(`/api/tasks?limit=${a.limit ?? 20}${a.state ? `&state=${encodeURIComponent(a.state)}` : ''}${a.unit ? `&unit=${encodeURIComponent(a.unit)}` : ''}`), - }, { name: 'td_claim_task', - description: 'Claim one specific task by id. Without an id it claims the next group (like td_claim_batch) and returns the first task context plus the others in alsoClaimed. Returns the full task context: symbol ID and address, file to write, previous attempts and their feedback, reviewer comments.', - inputSchema: { type: 'object', properties: { taskId: int('task to claim; omit for the next available') } }, + description: 'Claim one function that came back (needs_cleanup) to fix it, by task id. New work is claimed by region (td_regions, td_claim_region). Returns the task context: symbol, file, previous attempts and their feedback, reviewer comments.', + inputSchema: { type: 'object', properties: { taskId: int('the task to fix') }, required: ['taskId'] }, run: async (ws, a) => ops.api(ws).post('/api/tasks/claim', { taskId: a.taskId }), }, { @@ -127,12 +121,6 @@ const TOOLS = [ }, run: async (ws, a) => ops.submit(ws, a.taskIds ?? (a.taskId ? [a.taskId] : []), { files: a.files, notes: a.notes, agentMeta: { client: 'td-mcp' } }), }, - { - name: 'td_claim_batch', - description: 'Claim many functions at once. With no arguments (the usual way): the next coherent group the server picks, a whole source file, a class, or neighbouring functions in the binary. Or every open function of a given source file (unit) or address range. Work through them, td_check, then td_submit them together. Much cheaper than one function at a time.', - inputSchema: { type: 'object', properties: { unit: str('repo-relative source file, e.g. Engine/Source/FrameXML/LoadXML.cpp'), start: str('range start, hex'), end: str('range end, hex (exclusive)'), limit: int('max functions, default 50') } }, - run: async (ws, a) => ops.claimBatch(ws, a), - }, { name: 'td_mark_no_decompile', description: diff --git a/src/client/ops.js b/src/client/ops.js index 41e55c0..a893dee 100644 --- a/src/client/ops.js +++ b/src/client/ops.js @@ -229,8 +229,8 @@ export async function original(ws, query, kind) { /** Submits files for a leased task. Defaults to all changed files the task allows. */ /** - * Submits work. `taskIds` empty = every task you hold, as one batch (a whole unit you claimed - * with claim-batch, or several single tasks). Defaults to all changed files those tasks allow. + * Submits work. `taskIds` empty = every task you hold, as one batch (the regions you claimed, + * plus functions you took back to fix). Defaults to all changed files those tasks allow. */ export async function submit(ws, taskIds = [], { files, notes, agentMeta } = {}) { const me = await api(ws).get('/api/me'); @@ -340,10 +340,6 @@ export async function toolchainInstall(ws, { log = console.log } = {}) { }; } -/** Claims every open function of a unit (source file) or an address range, as one batch. */ -export function claimBatch(ws, { unit, start, end, limit } = {}) { - return api(ws).post('/api/tasks/claim-batch', { unit, start, end, limit }); -} export function relPath(ws, f) { return path.relative(ws.dir, path.resolve(f)).replace(/\\/g, '/'); diff --git a/src/server/app.js b/src/server/app.js index a7510ca..c563686 100644 --- a/src/server/app.js +++ b/src/server/app.js @@ -250,7 +250,7 @@ export function createApp(ctx) { })); router.on('GET', '/api/progress', () => progressData()); - const progressData = () => cached('progress', 10_000, () => { + const progressData = () => cached('progress', 20_000, () => { // Functions accepted as needing no decompilation (thunks, runtime library, ...) do not count. const total = store.get(`SELECT COUNT(*) AS n, COALESCE(SUM(a.size), 0) AS bytes FROM symbols s JOIN symbol_addresses a ON a.symbol_id = s.id JOIN versions v ON v.id = a.version_id AND v.is_current = 1 WHERE s.kind = 'function' AND s.library = 0 AND s.skip_reason IS NULL`); const byStatus = store.all(`SELECT fs.status, COUNT(*) AS n, COALESCE(SUM(a.size), 0) AS bytes FROM func_status fs JOIN symbol_addresses a ON a.symbol_id = fs.symbol_id JOIN versions v ON v.id = a.version_id AND v.is_current = 1 GROUP BY fs.status`); @@ -260,7 +260,16 @@ export function createApp(ctx) { const peerQueue = store.get("SELECT COUNT(*) AS n FROM attempts WHERE state = 'peer_review'").n; // Functions merged into main per hour: the last hour, and the average over the last day. const mergedSince = (ms) => store.get(`SELECT COALESCE(SUM(json_extract(data, '$.functions')), 0) AS n FROM events WHERE type = 'merged' AND at > ?`, Date.now() - ms).n; - const merges = { lastHour: mergedSince(3600_000), perHourDay: mergedSince(24 * 3600_000) / 24 }; + // Bytes of the original those merges cover (the functions of each merged batch). + const bytesSince = (ms) => + store.get( + // The merged batches first, so the attempts are found by batch (an index), not by scanning. + `WITH b AS MATERIALIZED (SELECT DISTINCT json_extract(data, '$.batch') AS id FROM events WHERE type = 'merged' AND at > ?) + SELECT COALESCE(SUM(sa.size), 0) AS n FROM b JOIN attempts at ON at.batch_id = b.id AND at.state = 'merged' + JOIN tasks t ON t.id = at.task_id JOIN symbol_addresses sa ON sa.symbol_id = t.symbol_id JOIN versions v ON v.id = sa.version_id AND v.is_current = 1`, + Date.now() - ms, + ).n; + const merges = { lastHour: mergedSince(3600_000), perHourDay: mergedSince(24 * 3600_000) / 24, bytesLastHour: bytesSince(3600_000), bytesPerHourDay: bytesSince(24 * 3600_000) / 24 }; return { functions: total.n, bytes: total.bytes, byStatus, confirmed, peerQueue, byTaskState, contributors, merges }; }); @@ -380,26 +389,25 @@ export function createApp(ctx) { const user = auth(req); const body = await readJson(req, 4096); if (body.taskId) { - // Taking back one's own returned function is not new work. - const own = store.get('SELECT 1 FROM tasks WHERE id = ? AND preferred_user = ? AND preferred_until > ?', intParam(body.taskId), user.id, Date.now()); + // Single functions only to fix one that came back; new work is claimed by region. + const t = tasks.get(intParam(body.taskId)); + if (!['needs_cleanup', 'hard'].includes(t.state)) throw new HttpError(409, 'claim new work by region: td_regions, then td_claim_region (a single function can only be claimed to fix one that came back)'); + const own = store.get('SELECT 1 FROM tasks WHERE id = ? AND preferred_user = ? AND preferred_until > ?', t.id, user.id, Date.now()); if (!own) reviewFirst(user); - return guided(tasks.context(tasks.claim(user, intParam(body.taskId)))); + return guided(tasks.context(tasks.claim(user, t.id))); } reviewFirst(user); - // "The next task" is the next group: one function at a time across classes wastes the - // agent's context and the compile per submission. - const ids = tasks.claimNext(user, 50); - const others = ids.slice(1).map((id) => { - const t = tasks.get(id); - return { task: id, symbol: t.symbol_id }; - }); - return guided({ - ...tasks.context(ids[0]), - alsoClaimed: others, - note: groupNote(ids.length, tasks.lastGroup), - }); + // "The next work" is the best free region, as a whole. + return guided(claimBestRegion(user)); }); + const claimBestRegion = (user) => { + const pick = ctx.regions.list({ user, free: true, limit: 1 }).regions[0] ?? ctx.regions.list({ user, limit: 1 }).regions[0]; + if (!pick) throw new HttpError(404, 'no open regions'); + const r = ctx.regions.claim(user, { kind: 'unit', key: pick.key }); + return { ...r, baseCommit: repo.head(), note: 'Claimed the best free region for you. Next time pick one yourself: td_regions, then td_claim_region.' }; + }; + // "No decompile": held functions claimed to need no source; peer reviewed like code. router.on('POST', '/api/tasks/no-decompile', async ({ req }) => { const user = auth(req); @@ -470,8 +478,12 @@ export function createApp(ctx) { const user = auth(req); const body = await readJson(req, 4096); reviewFirst(user); - const hexOrNull = (x) => (x == null ? null : typeof x === 'string' ? Number.parseInt(x, 16) : Number(x)); - const ids = tasks.claimBatch(user, { unit: body.unit ? String(body.unit) : null, start: hexOrNull(body.start), end: hexOrNull(body.end), limit: body.limit ? Number(body.limit) : 50 }); + // Retired: work is claimed by region. A file becomes its region; nothing else is handed out. + if (body.start != null || body.end != null) throw new HttpError(410, 'address ranges are retired: claim a region (td_regions, then td_claim_region)'); + if (!body.unit) return guided(claimBestRegion(user)); + const claimed = ctx.regions.claim(user, { kind: 'unit', key: String(body.unit) }); + const ids = claimed.leased; + if (!ids.length) return guided({ ...claimed, baseCommit: repo.head(), tasks: [], note: 'nothing open to lease in that file (it may be shared: you can still submit candidates for its functions)' }); const v = store.currentVersion(); const ph = ids.map(() => '?').join(','); const rows = store.all( @@ -483,8 +495,9 @@ export function createApp(ctx) { tasks: rows.map((r) => ({ ...r, addr: r.addr != null ? '0x' + hex(r.addr) : null, annotation: `// FUNCTION: ${r.symbol}` })), allowedPaths: tasks.allowedPathsForTasks(ids), baseCommit: repo.head(), - group: body.unit || body.start != null ? undefined : tasks.lastGroup && { kind: tasks.lastGroup.kind, key: tasks.lastGroup.key, note: groupNote(ids.length, tasks.lastGroup) }, - note: 'Write or fix every function, check, then submit them together (td submit --all). Context of one task: GET /api/tasks/.', + region: claimed.region, + alsoClaimedBy: claimed.alsoClaimedBy, + note: 'Write or fix every function, check, then submit them together (td submit). Context of one task: GET /api/tasks/.', }); }); diff --git a/src/server/db.js b/src/server/db.js index 8967850..bf6f482 100644 --- a/src/server/db.js +++ b/src/server/db.js @@ -307,6 +307,7 @@ export class Store { this.addColumn('attempts', 'batch_id', 'INTEGER REFERENCES batches(id)'); // A candidate for a function another agent holds (region claims): it does not move the task. this.addColumn('attempts', 'competing', 'INTEGER NOT NULL DEFAULT 0'); + this.db.exec('CREATE INDEX IF NOT EXISTS events_type_at ON events(type, at)'); // merge rates, feed this.db.exec('CREATE INDEX IF NOT EXISTS attempts_batch ON attempts(batch_id)'); } diff --git a/src/server/web/app.js b/src/server/web/app.js index 69d61cd..d57358b 100644 --- a/src/server/web/app.js +++ b/src/server/web/app.js @@ -351,6 +351,20 @@ function renderProgress(p) { row.append(right); bars.append(row); }; + // The pace: what landed on main in the last hour, and the average over the last day. + if (p.merges) { + const rate = el('div', null, 'rates'); + const r = (n, unit, avg) => { + const box = el('div', null, 'rate'); + box.append(el('span', n, 'rate-n'), el('span', unit, 'rate-u'), el('span', avg, 'muted rate-avg')); + return box; + }; + rate.append( + r(fmt(p.merges.lastHour), 'functions / h', `avg ${p.merges.perHourDay.toFixed(1)}/h over 24 h`), + r(kb(p.merges.bytesLastHour ?? 0), 'of code / h', `avg ${kb(p.merges.bytesPerHourDay ?? 0)}/h over 24 h`), + ); + bars.append(rate); + } bar('fn', 'Functions', { identical: identical.n, confirmed: confirmed.n }, p.functions, fmt); bar('by', 'Code', { identical: identical.bytes, confirmed: confirmed.bytes }, p.bytes, kb); const legend = el('div', null, 'pbar-legend'); diff --git a/src/server/web/style.css b/src/server/web/style.css index de3f3c5..3f2b7b7 100644 --- a/src/server/web/style.css +++ b/src/server/web/style.css @@ -735,3 +735,10 @@ details.fold-plate[open] > summary { margin-bottom: 12px; } .view, .feed li.new, .toast, .wallet-coin, .reel.spinning, .die.rolling, .credits-badge.bump, .stat-tile.tick .n, .spark { animation: none; } .big-coin { transition: none; } } + +/* Progress: the pace */ +.rates { display: flex; flex-wrap: wrap; gap: 10px 28px; padding-bottom: 6px; border-bottom: 1px dashed var(--line); } +.rate { display: flex; align-items: baseline; gap: 8px; flex-wrap: wrap; } +.rate-n { font: 700 26px/1 var(--font-mono); color: var(--accent-ink); font-variant-numeric: tabular-nums; } +.rate-u { font: 600 13px var(--font-display); text-transform: uppercase; letter-spacing: 0.1em; } +.rate-avg { font: 12px var(--font-mono); } diff --git a/test/server.test.js b/test/server.test.js index bc1b6be..3351567 100644 --- a/test/server.test.js +++ b/test/server.test.js @@ -77,6 +77,15 @@ async function login(name, password = PASSWORD) { return r.setCookie.split(';')[0]; } +// One task leased directly: the API hands out new work only by region (single functions only to fix). +async function leaseTask(token, taskId) { + const me = (await call('GET', '/api/me', null, token)).body; + const user = ctx.accounts.byName(me.name); + const id = taskId ?? ctx.store.get(`SELECT id FROM tasks WHERE state IN ('open', 'needs_cleanup') ORDER BY id LIMIT 1`).id; + ctx.tasks.claim(user, id); + return { status: 200, body: ctx.tasks.context(id) }; +} + async function waitFor(fn, ms = 5000) { const end = Date.now() + ms; for (;;) { @@ -120,7 +129,7 @@ test('claim and static-gate rejection', async () => { assert.equal((await call('GET', '/api/me', null, token)).body.name, 'alice'); assert.equal((await call('GET', '/api/me', null, 'td_bogus')).status, 401); - const claim = await call('POST', '/api/tasks/claim', {}, token); + const claim = await leaseTask(token); assert.equal(claim.status, 200); const ctxBody = claim.body; // "Next" claims a whole group; this test works on one task. @@ -211,7 +220,7 @@ test('regressions are detected against func_status', () => { test('expired leases return to the pool', async () => { const token = await activeUser('dave'); - const c = (await call('POST', '/api/tasks/claim', {}, token)).body; + const c = (await leaseTask(token)).body; for (const o of c.alsoClaimed ?? []) await call('POST', `/api/tasks/${o.task}/release`, {}, token); ctx.store.run('UPDATE tasks SET lease_expires = 1 WHERE id = ?', c.task.id); assert.equal(ctx.tasks.sweepExpired(), 1); @@ -267,7 +276,7 @@ test('API tokens: created in the browser, limited scope, revocable', async () => test('revoking an account kills its tokens, sessions and leases', async () => { const token = await activeUser('frank'); const session = await login('frank'); - const claim = await call('POST', '/api/tasks/claim', {}, token); + const claim = await leaseTask(token); assert.equal(claim.status, 200); const admin = await login('root'); @@ -505,7 +514,7 @@ test('agents place held, unplaced functions into files before porting them', asy ctx.store.run('UPDATE symbols SET unit = NULL, name = NULL WHERE id = ?', id); ctx.tasks.generate(); const task = ctx.store.get('SELECT id FROM tasks WHERE symbol_id = ?', id).id; - assert.equal((await call('POST', '/api/tasks/claim', { taskId: task }, token)).status, 200); + assert.equal((await leaseTask(token, task)).status, 200); const bad = await call('POST', '/api/tasks/place', { assignments: [{ taskId: task, unit: '../etc/x.cpp' }] }, token); assert.equal(bad.status, 400); @@ -665,7 +674,7 @@ test('"no decompile" claims are peer reviewed; accepted ones are skipped and not const real = mk(0x411040); ctx.tasks.generate(); const taskOf = (sym) => ctx.store.get('SELECT id FROM tasks WHERE symbol_id = ?', sym).id; - for (const sym of [thunk, real]) assert.equal((await call('POST', '/api/tasks/claim', { taskId: taskOf(sym) }, author)).status, 200); + for (const sym of [thunk, real]) assert.equal((await leaseTask(author, taskOf(sym))).status, 200); const bad = await call('POST', '/api/tasks/no-decompile', { items: [{ taskId: taskOf(thunk), category: 'whatever', reason: 'it is a thunk, add rcx then jmp' }] }, author); assert.equal(bad.status, 400); @@ -712,7 +721,7 @@ test('verification: unwritten functions stay claimed, failing ones go back alone const guarded = mk(0x412100); ctx.tasks.generate(); const taskOf = (sym) => ctx.store.get('SELECT id FROM tasks WHERE symbol_id = ?', sym).id; - for (const sym of [good, broken, unwritten]) assert.equal((await call('POST', '/api/tasks/claim', { taskId: taskOf(sym) }, author)).status, 200); + for (const sym of [good, broken, unwritten]) assert.equal((await leaseTask(author, taskOf(sym))).status, 200); const fn = (id, status, hash) => ({ id, status, score: status === STATUS.EXACT ? 1 : 0.5, file: 'src/pv.cpp', codeHash: hash, relocations: [], size: { ours: 8, original: 8 } }); let guardedHash = 'g1'; const real = ctx.verifier.check; @@ -741,7 +750,7 @@ test('verification: unwritten functions stay claimed, failing ones go back alone // Changing a function another account holds is still refused. ctx.tasks.generate(); - assert.equal((await call('POST', '/api/tasks/claim', { taskId: taskOf(guarded) }, other)).status, 200); + assert.equal((await leaseTask(other, taskOf(guarded))).status, 200); ctx.store.run('INSERT OR REPLACE INTO func_status (symbol_id, file, status, score, commit_hash, updated_at, code_hash) VALUES (?, ?, ?, 0.5, ?, 0, ?)', guarded, 'src/pv.cpp', 'mismatch', 'x', 'g0'); const sub2 = await call('POST', '/api/batches', { taskIds: [taskOf(unwritten)], baseCommit: ctx.repo.head(), files: { 'src/pv.cpp': `// FUNCTION: ${unwritten}\nint uw() { return 1; }\n// FUNCTION: ${guarded}\nint gd() { return 4; }\n` } }, author); await waitFor(async () => !['queued', 'verifying'].includes(ctx.store.get('SELECT state FROM batches WHERE id = ?', sub2.body.batchId).state)); @@ -1008,7 +1017,7 @@ test('batches: one compile, auto-approval, peer review, one function sent back, bHash = 'hash-b-2'; const objector = await activeUser('bpeer6'); const taskOfB = claim.body.tasks.find((t) => t.symbol === b).task; - assert.equal((await call('POST', '/api/tasks/claim', { taskId: taskOfB }, author)).status, 200); + assert.equal((await leaseTask(author, taskOfB)).status, 200); const s2 = await call('POST', '/api/batches', { taskIds: [taskOfB], baseCommit: ctx.repo.head(), files: { 'src/batch.cpp': source } }, author); const r2 = Object.fromEntries((await settle(s2.body.batchId)).functions.map((f) => [f.symbol, f])); assert.equal(r2[b].state, 'peer_review', 'changed code gets reviewed again'); @@ -1113,6 +1122,13 @@ test('regions: two agents claim one file, the first accepted candidate merges, t ctx.verifier.check = realCheck; }); +test('new work is claimed by region; a single function only to fix one that came back', async () => { + const token = await activeUser('rtuser'); + const open = ctx.store.get(`SELECT id FROM tasks WHERE state = 'open' LIMIT 1`); + if (open) assert.equal((await call('POST', '/api/tasks/claim', { taskId: open.id }, token)).status, 409); + assert.equal((await call('POST', '/api/tasks/claim-batch', { start: '0x401000', end: '0x402000' }, token)).status, 410); +}); + test('the server does not compile checks unless remoteChecks is on', async () => { const user = await activeUser('rcuser'); ctx.config.remoteChecks = false;