Skip to content

fix(task-board): fix session-task linkage and stuck workflow advancement #436

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dimakis merged 8 commits into from
Jul 4, 2026
Merged

fix(task-board): fix session-task linkage and stuck workflow advancement #436

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
83a9bb8
fix(task-board): fix session-task linkage and stuck workflow advancement
dimakis Jul 4, 2026
04175c9
fix(task-board): address Centaur review findings
dimakis Jul 4, 2026
18304a9
fix(task-board): address second Centaur review — resume paused orches…
dimakis Jul 4, 2026
506afc4
fix(task-board): address Centaur R3 findings
dimakis Jul 4, 2026
a3fabae
fix(task-board): address third Centaur review — keys() helper, idle l…
dimakis Jul 4, 2026
bdfe0bc
fix(task-board): address Centaur review — goal-change test, style fix
dimakis Jul 4, 2026
f9ab95c
fix(task-board): fix TS null check in .finally() handler
dimakis Jul 4, 2026
af9a738
fix(task-board): address Centaur R2 — try-catch in .finally(), test h…
dimakis Jul 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions packages/harness/src/session-registry.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,10 @@ export class SessionRegistry {
return this.sessions.get(clientId);
}

keys(): IterableIterator<string> {
return this.sessions.keys();
}

entries(): IterableIterator<[string, ManagedSession]> {
return this.sessions.entries();
}
Expand Down
185 changes: 179 additions & 6 deletions server/__tests__/task-orchestrator.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -312,18 +312,19 @@ describe('TaskOrchestrator', () => {

describe('orphan detection', () => {
it('reclaims orphaned tasks during tick', () => {
// Create deps with getActiveSessionIds
// getActiveSessionIds returns clientIds (not SDK sessionIds)
// to match what setSessionId stores on tasks
const depsWithOrphan = createTestDeps(store);
depsWithOrphan.getActiveSessionIds = () => new Set(['alive-session']);
depsWithOrphan.getActiveSessionIds = () => new Set(['alive-client']);
const orch = new TaskOrchestrator(depsWithOrphan);

const goal = store.create({ title: 'Goal' });
const c1 = store.create({ title: 'Orphan', parentId: goal.id });
store.create({ title: 'Next', parentId: goal.id });

// Simulate c1 assigned to dead session
// Simulate c1 assigned to dead session (clientId not in active set)
store.update(c1.id, { status: 'active' });
store.setSessionId(c1.id, 'dead-session');
store.setSessionId(c1.id, 'dead-client');

orch.start(goal.id);

Expand All @@ -335,21 +336,50 @@ describe('TaskOrchestrator', () => {

it('does not reclaim tasks with alive sessions', () => {
const depsWithOrphan = createTestDeps(store);
depsWithOrphan.getActiveSessionIds = () => new Set(['alive-session']);
depsWithOrphan.getActiveSessionIds = () => new Set(['alive-client']);
const orch = new TaskOrchestrator(depsWithOrphan);

const goal = store.create({ title: 'Goal' });
const c1 = store.create({ title: 'Active', parentId: goal.id });
const c2 = store.create({ title: 'Next', parentId: goal.id });

store.update(c1.id, { status: 'active' });
store.setSessionId(c1.id, 'alive-session');
store.setSessionId(c1.id, 'alive-client');

orch.start(goal.id);

// c1 is alive, so tick should skip it and pick c2
expect(orch.getStatus().activeTaskId).toBe(c2.id);
});

it('reclaims spawned task sessions using clientId matching', () => {
// Simulates the real scenario: task.session_id stores clientId
// (e.g. 'task:abc123'), active set contains clientIds from registry
const depsWithOrphan = createTestDeps(store);
depsWithOrphan.getActiveSessionIds = () => new Set(['task:alive-wt']);
const orch = new TaskOrchestrator(depsWithOrphan);

const goal = store.create({ title: 'Goal' });
const orphan = store.create({ title: 'Dead spawn', parentId: goal.id });
const alive = store.create({ title: 'Alive spawn', parentId: goal.id });
store.create({ title: 'Pending', parentId: goal.id });

// Orphan: session ended, clientId no longer in registry
store.update(orphan.id, { status: 'active' });
store.setSessionId(orphan.id, 'task:dead-wt');

// Alive: session still running
store.update(alive.id, { status: 'active' });
store.setSessionId(alive.id, 'task:alive-wt');

orch.start(goal.id);

// Orphan should be reclaimed and re-dispatched
expect(store.get(orphan.id)!.sessionId).toBeNull();
// Alive should NOT be reclaimed
expect(store.get(alive.id)!.sessionId).toBe('task:alive-wt');
expect(store.get(alive.id)!.status).toBe('active');
});
});

describe('task approval', () => {
Expand Down Expand Up @@ -882,6 +912,149 @@ describe('TaskOrchestrator', () => {
});
});

it('tick after spawn session completes advances workflow', async () => {
const spawnSession = vi.fn().mockResolvedValue('task:spawned-1');
const deps = createTestDeps(store);
deps.spawnSession = spawnSession;
let activeClients = new Set(['task:spawned-1']);
deps.getActiveSessionIds = () => activeClients;
const orch = new TaskOrchestrator(deps);

const goal = store.create({ title: 'Goal' });
const t1 = store.create({ title: 'Spawn task', parentId: goal.id });
const t2 = store.create({ title: 'Next task', parentId: goal.id, sessionPolicy: 'reuse' });

orch.start(goal.id);

await vi.waitFor(() => {
expect(spawnSession).toHaveBeenCalled();
});

// Agent completed the task during the session
store.update(t1.id, { status: 'done' });
store.cascadeStatus(t1.id);

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 missing_tests: The 'tick after spawn session completes advances workflow' test doesn't assert that spawnSession was called exactly once. Adding expect(spawnSession).toHaveBeenCalledTimes(1) after verifying t2 is active would prove t2 used the reuse path (not spawn), which is the key workflow-advancement contract being tested. [fixable]


// .finally() fires: session removed from registry, tick() called directly
activeClients = new Set();
orch.tick();

// t2 should now be active (workflow advanced)
expect(store.get(t2.id)!.status).toBe('active');
expect(orch.getStatus().activeTaskId).toBe(t2.id);
});

it('resume after spawn session dies reclaims unfinished task', async () => {

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 missing_tests: The "resume after spawn session dies" test covers the crash → reclaim → re-dispatch path well, but doesn't verify the reclaimed task's status transitions (active → pending → active). Adding expect(store.get(task.id)!.status).toBe('pending') before the resume() call would document the pre-condition more clearly and guard against store-level regressions. [fixable]

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 missing_tests: The 'resume after spawn session dies' test relies on the .then() callback from the mock spawnSession setting sessionId on the task (via queueMicrotask flushing during vi.waitFor). This implicit dependency makes the test fragile — if vi.waitFor timing changes, the sessionId might not be set when resume() runs, causing orphan detection to miss the task (no session_id = no orphan). Consider adding an explicit assertion that task.sessionId is set before simulating the crash, e.g. expect(store.get(task.id)!.sessionId).toBe('task:spawned-1') before clearing activeClients. [fixable]

// Primary scenario for .finally(): session crashes without calling
// TaskComplete. The task stays active but the session is gone.
// After all tasks are spawned, the orchestrator pauses (no pending left).
// .finally() calls resume() → tick() → orphan reclaim → re-dispatch.
const spawnSession = vi.fn().mockResolvedValue('task:spawned-1');
const deps = createTestDeps(store);
deps.spawnSession = spawnSession;
let activeClients = new Set(['task:spawned-1']);
deps.getActiveSessionIds = () => activeClients;
const orch = new TaskOrchestrator(deps);

const goal = store.create({ title: 'Goal' });
store.create({ title: 'Spawn task', parentId: goal.id });

orch.start(goal.id);

await vi.waitFor(() => {
expect(spawnSession).toHaveBeenCalledTimes(1);
});

// Orchestrator should be paused (all tasks active, none pending)
expect(orch.getStatus().state).toBe('paused');

// Verify the task is active and has sessionId set (pre-condition for orphan reclaim)
const task = store.getChildren(goal.id)[0];
expect(store.get(task.id)!.status).toBe('active');
expect(store.get(task.id)!.sessionId).toBe('task:spawned-1');

// .finally() fires: session removed from registry, resume() called
activeClients = new Set();
orch.resume();

// Orphan reclaim detected the dead session and re-dispatched the task
await vi.waitFor(() => {
expect(spawnSession).toHaveBeenCalledTimes(2);
});
});

it('tick is a no-op when orchestrator is stopped before session ends', async () => {
// Edge case: orchestrator stopped (idle) while a spawned session is
// still in-flight. .finally() fires → tick() → returns early (guarded
// by state !== 'running'). No crash, no side effects.
const spawnSession = vi.fn().mockResolvedValue('task:spawned-1');
const deps = createTestDeps(store);
deps.spawnSession = spawnSession;
// Return spawned clientId while session is alive
let activeClients = new Set(['task:spawned-1']);
deps.getActiveSessionIds = () => activeClients;
const orch = new TaskOrchestrator(deps);

const goal = store.create({ title: 'Goal' });
const t1 = store.create({ title: 'Spawn task', parentId: goal.id });

orch.start(goal.id);

await vi.waitFor(() => {
expect(spawnSession).toHaveBeenCalledTimes(1);
});

// User stops the orchestrator while session is still running
orch.stop();
expect(orch.getStatus().state).toBe('idle');

// .finally() fires after session ends — tick is a no-op in idle state
activeClients = new Set();
orch.tick();

expect(orch.getStatus().state).toBe('idle');
expect(store.get(t1.id)?.status).toBe('active'); // unchanged
expect(spawnSession).toHaveBeenCalledTimes(1); // no re-dispatch
});

it('does not advance old goal when orchestrator moved to a new goal', async () => {
const spawnSession = vi.fn().mockResolvedValue('task:spawned-1');
const deps = createTestDeps(store);
deps.spawnSession = spawnSession;
let activeClients = new Set(['task:spawned-1']);
deps.getActiveSessionIds = () => activeClients;
const orch = new TaskOrchestrator(deps);

const goal1 = store.create({ title: 'Goal 1' });
store.create({ title: 'Spawn task', parentId: goal1.id });

const goal2 = store.create({ title: 'Goal 2' });
const g2task = store.create({
title: 'Reuse task',
parentId: goal2.id,
sessionPolicy: 'reuse',
});

orch.start(goal1.id);

await vi.waitFor(() => {
expect(spawnSession).toHaveBeenCalledTimes(1);
});

// User switches to a new goal while spawned session is still running
orch.stop();
orch.start(goal2.id);
expect(orch.getStatus().goalId).toBe(goal2.id);
expect(orch.getStatus().activeTaskId).toBe(g2task.id);

// Old spawned session ends — tick should NOT interfere with goal2
activeClients = new Set();
orch.tick();

// Goal2 state should be unchanged
expect(orch.getStatus().goalId).toBe(goal2.id);
expect(orch.getStatus().activeTaskId).toBe(g2task.id);
});

it('reuse policy tasks use pinned session as before', () => {
const spawnSession = vi.fn().mockResolvedValue('spawned-client-1');
const deps = createTestDeps(store);
Expand Down
58 changes: 44 additions & 14 deletions server/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,15 +235,12 @@ const orchestrator = new TaskOrchestrator({
sseRegistry.broadcast('task_state', data);
},
getActiveSessionIds: () => {
const ids = new Set<string>();
for (const [clientId] of registry.entries()) {
const session = registry.get(clientId);
if (session?.sessionId) ids.add(session.sessionId);
}
return ids;
// Return clientIds — task.session_id stores clientId, not SDK sessionId.
// Using sessionId here caused orphan detection to never match spawned tasks.

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 style: The loop for (const [clientId] of registry.entries()) { ids.add(clientId); } destructures tuples just to get the key. If SessionRegistry exposes a keys() iterator or a size/clientIds accessor, that would be cleaner. Minor — current form is clear enough. [fixable]

return new Set(registry.keys());

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 unsafe_assumptions: The new getActiveSessionIds returns ALL clientIds from the registry via registry.keys(), including regular browser-connected sessions. Previously it only returned SDK sessionIds from active sessions. This means orphan detection now compares task session_id (which stores clientIds like task:abc123) against ALL registry keys (which also include browser client IDs). This works correctly because browser clientIds will never match a task:* sessionId stored on a task, but it's a broader set than strictly necessary. Not a bug — just a noted semantic change.

},

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 style: Array.from(registry.entries(), ([clientId]) => clientId) allocates an intermediate array just to feed it into a Set. A clearer alternative: new Set(registry.keys()) if SessionRegistry exposes a keys() method, or the existing loop pattern (for...of with ids.add) which avoids the allocation. [fixable]

spawnSession: async (taskId: string, prompt: string, goalId: string) => {
const clientId = `headless:${generateWtId()}`;
const clientId = `task:${generateWtId()}`;

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 style: The task: prefix for spawned sessions is a good improvement over headless:, but app.ts:501 still uses headless: for externally-created headless sessions (POST /api/sessions). The inconsistency is intentional (different use cases), but worth noting that both code paths create NullTransport sessions — only the task path gets .finally() cleanup. If external headless sessions also need orphan-aware cleanup in the future, the same pattern would be needed there.

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 style: The prefix was renamed from headless: to task: which is clearer. However, log.info on line 256 still says 'task session resolved' and line 261 says 'task session failed' — these are fine. But the catch block at line 306 still says 'failed to spawn session for task', which could be confused with line 261's 'task session failed' during log searching. Consider differentiating: one is the async session failure (agent error) and the other is the synchronous spawn failure (e.g., transport creation). Minor log clarity issue. [fixable]

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 style: The prefix changed from 'headless:' to 'task:' which is a clearer name. However, no migration handling exists for tasks in SQLite that may still reference 'headless:' session_ids from before this change. On server restart, such tasks would be orphaned and reclaimed (correct behavior), but worth noting in the commit message that this is intentional — the prefix change applies only to new sessions.

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 style: Prefix changed from headless: to task: — this is a good semantic improvement, but any external logging/monitoring/alerting that greps for headless: will silently stop matching. If there are log queries or dashboards filtering on the old prefix, they should be updated. Not a code issue, but worth noting in the PR description.


try {
const transport = new NullTransport();
Expand All @@ -257,16 +254,49 @@ const orchestrator = new TaskOrchestrator({
telosTaskId: goalId,
taskContext: { currentTaskId: taskId, goalId },
onSessionResolved: (sessionId) => {
log.info('spawned headless session resolved', { taskId, sessionId, clientId });
log.info('task session resolved', { taskId, sessionId, clientId });
sseRegistry.broadcast('sessions_changed', {});
},
}).catch((err) => {
log.error('spawned session failed', {
taskId,
clientId,
error: (err as Error).message,
})
.catch((err) => {
log.error('task session failed', {
taskId,
clientId,
error: (err as Error).message,
});
})
.finally(() => {

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 style: The .finally() block is 22 lines with branching logic (paused vs running, goal match vs mismatch). Consider extracting this into a named function (e.g., onTaskSessionEnded(clientId, taskId, goalId)) to improve readability and make the intent self-documenting. The current inline approach works but is dense for a fire-and-forget chain. [fixable]

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 bugs: The .finally() handler calls orchestratorRef.tick() or .resume() without a try-catch. These methods call into TaskStore (SQLite queries) which can throw on database errors. Since .finally() is on a promise chain, an exception here becomes an unhandled promise rejection — the session cleanup (registry.remove) has already happened, but the error is silently swallowed in production or crashes the process depending on Node.js --unhandled-rejections mode. Wrap the orchestrator calls in try-catch with a log.error fallback. [fixable]

// Session ended (success or failure) — clean up registry entry
// (no WS close to trigger normal removal) and advance the
// orchestrator so orphan reclaim picks up unfinished tasks.
registry.remove(clientId);
// Only advance if the orchestrator is still on the same goal.
// A user may have paused manually or started a new goal while
// this session was in-flight — don't override that.
if (!orchestratorRef) return;
try {
const status = orchestratorRef.getStatus();
if (status.goalId === goalId) {
if (status.state === 'paused') {
orchestratorRef.resume();
} else {
orchestratorRef.tick();
}
} else {
log.info('task session ended after loop stopped or goal changed', {

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 bugs: The .finally() handler calls resume() whenever state === 'paused', but cannot distinguish between auto-pause (all tasks dispatched — correct to resume) and user-initiated pause() (should stay paused). If a user manually pauses the orchestrator while spawned sessions are in-flight, the next session to end will call resume() and override the user's intent. Consider tracking a manuallyPaused flag on the orchestrator, or changing the .finally() to always call tick() — and having tick() internally handle orphan reclaim even in paused state. [fixable]

taskId,
clientId,
goalId,
});
}
} catch (err: unknown) {
log.error('orchestrator advance failed after task session end', {
taskId,
clientId,
error: err instanceof Error ? err.message : String(err),
});
}
});
});

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 unsafe_assumptions: The comment 'register() is synchronous inside startChat, so the session exists here' is correct today (registry.register at chat.ts:848 runs before any await in _startChatInner), but this is a fragile implicit contract. If a future refactor adds an await before register() (e.g. async validation), the registry.get(clientId) here would silently return undefined and skip the taskContext assignment. Consider moving the taskContext assignment into the spawnSession callback after setSessionId in task-orchestrator.ts, or documenting the contract at the register() call site. [fixable]

return clientId;
} catch (err) {
Expand Down
Loading