diff --git a/packages/agent-connector/src/adapters/base.js b/packages/agent-connector/src/adapters/base.js index 8d931a3bb..9cd177868 100644 --- a/packages/agent-connector/src/adapters/base.js +++ b/packages/agent-connector/src/adapters/base.js @@ -19,6 +19,7 @@ const { WorkspaceClient, SessionRevokedError } = require('../workspace-client'); const { generateSessionTitle, SESSION_DEFAULT_RE } = require('./utils'); +const { buildPhaseGateDirective } = require('./workspace-prompt'); const { defaultAgentWorkdir } = require('../paths'); const { REASON, @@ -75,6 +76,11 @@ class BaseAdapter { this._processedIds = new Set(); this._titledSessions = new Set(); this._mode = 'execute'; + // Per-channel, per-message mode override set by the clarification phase + // gate (see _modeFor). Keyed by channel — the same channel is processed + // serially, but different channels run in parallel, so a single scalar + // would leak one channel's gate onto another. + this._modeOverrides = {}; this._lastControlId = null; this._controlWake = null; // Per-channel task tracking for parallel execution @@ -699,7 +705,53 @@ class BaseAdapter { // Channel dispatch // ------------------------------------------------------------------ + /** + * Role this agent plays in an active clarification phase, from the routed + * message's metadata. Returns null when the channel isn't gated. + * + * 'plan' → the backend downgraded this wake-up (target_modes): answer, + * don't build + * 'owner' → this agent owns the phase and is the one who advances it + * 'member'→ phase is active, this agent is neither of the above + */ + _phaseRole(msg) { + const meta = (msg && msg.metadata) || {}; + if (meta.phase !== 'clarifying') return null; + const modes = meta.target_modes || {}; + if (modes[this.agentName] === 'plan') return 'plan'; + if (meta.phase_owner && meta.phase_owner === this.agentName) return 'owner'; + return 'member'; + } + + /** + * Append the clarification-phase directive to a routed message. + * + * The backend gate decides who may be woken; this is what stops a woken + * builder from implementing against an unsettled spec. Done here rather + * than per adapter so every adapter type is covered by one code path, and + * appended (not prepended) so channel auto-titling still sees the user's + * own words first. + */ + _applyPhaseGate(msg) { + const role = this._phaseRole(msg); + if (!role) return msg; + const directive = buildPhaseGateDirective({ + role, + owner: (msg.metadata || {}).phase_owner, + endpoint: this.endpoint, + workspaceId: this.workspaceId, + channelName: msg.sessionId || this.channelName, + }); + if (!directive) return msg; + this._log(`Phase gate active (${role}) for message ${msg.messageId || '?'}`); + return { ...msg, content: `${msg.content || ''}${directive}` }; + } + async _dispatchMessage(msg) { + // Carry the phase directive on the message itself so a queued message + // still holds the constraint it arrived under when it is finally run. + msg = this._applyPhaseGate(msg); + // Use sessionId only if it looks like a channel, not an agent target let channel = this.channelName || 'general'; if (msg.sessionId && !msg.sessionId.startsWith('openagents:') && !msg.sessionId.startsWith('agent:')) { @@ -744,10 +796,39 @@ class BaseAdapter { return true; } + /** + * The mode this agent must run in for work on `channel` right now: the + * agent's own mode, unless the message being handled was gated into PLAN + * by the clarification phase. Adapters that enforce plan mode at the + * runtime level (read-only tools, no writes) should read this instead of + * `this._mode` so a gated wake-up genuinely cannot build. + */ + _modeFor(channel) { + return this._modeOverrides[channel] || this._mode; + } + + /** + * Run one message with its phase-gate mode override in effect. The + * override is per channel and cleared afterwards, so it applies to exactly + * the message it arrived with. + */ + async _runMessage(channel, msg) { + if (this._phaseRole(msg) === 'plan') { + this._modeOverrides[channel] = 'plan'; + } else { + delete this._modeOverrides[channel]; + } + try { + await this._handleMessage(msg); + } finally { + delete this._modeOverrides[channel]; + } + } + async _channelWorker(channel, msg) { this._channelBusy.add(channel); try { - await this._handleMessage(msg); + await this._runMessage(channel, msg); } catch (e) { this._log(`Error in channel worker for ${channel}: ${e.message}`); try { await this.sendError(channel, `Agent error: ${e.message}`); } catch {} @@ -762,7 +843,7 @@ class BaseAdapter { try { await this.sendStatus(channel, 'processing queued message', { queue_id: nextMsg._queueId, queue_status: 'processed' }); } catch {} } try { - await this._handleMessage(nextMsg); + await this._runMessage(channel, nextMsg); } catch (e) { this._log(`Error processing queued message in ${channel}: ${e.message}`); try { await this.sendError(channel, `Agent error: ${e.message}`); } catch {} diff --git a/packages/agent-connector/src/adapters/claude.js b/packages/agent-connector/src/adapters/claude.js index aecdf0f5f..75e006367 100644 --- a/packages/agent-connector/src/adapters/claude.js +++ b/packages/agent-connector/src/adapters/claude.js @@ -489,7 +489,7 @@ class ClaudeAdapter extends BaseAdapter { agentName: this.agentName, workspaceId: this.workspaceId, channelName, - mode: this._mode, + mode: this._modeFor(channelName), browserEnabled, toolMode: this.toolMode, decisionLog, @@ -526,7 +526,7 @@ class ClaudeAdapter extends BaseAdapter { * Skills mode: write a SKILL.md file and allow Bash + curl for workspace ops. */ _buildSkillsCmd(cmd, channelName) { - if (this._mode === 'plan') { + if (this._modeFor(channelName) === 'plan') { cmd.push('--permission-mode', 'plan'); cmd.push('--allowedTools', 'Read', 'Glob', 'Grep', 'Bash'); } else { @@ -607,7 +607,7 @@ class ClaudeAdapter extends BaseAdapter { mcpTools.push(`${pfx}workspace_get_todos`, `${pfx}workspace_list_timers`, `${pfx}workspace_list_routines`); mcpWriteTools.push(`${pfx}workspace_put_todos`, `${pfx}workspace_create_timer`, `${pfx}workspace_cancel_timer`, `${pfx}workspace_create_routine`, `${pfx}workspace_cancel_routine`); - if (this._mode === 'plan') { + if (this._modeFor(channelName) === 'plan') { cmd.push('--permission-mode', 'plan'); cmd.push('--allowedTools', ...mcpTools, 'Read', 'Glob', 'Grep'); } else { @@ -1028,7 +1028,10 @@ class ClaudeAdapter extends BaseAdapter { * and the fresh-spawn path so their behavior can never drift. */ _composeFinalResponse(pp) { - if (this._mode === 'plan') { + // The spawn-time mode, not the current one: a phase-gated turn ran in + // plan mode even though the agent's own mode is execute, and its plan + // file is what the user must see. + if ((pp.spawnMode || this._mode) === 'plan') { try { const planDir = path.join(this.workingDir || defaultAgentWorkdir(this.agentName), '.claude', 'plans'); if (fs.existsSync(planDir)) { @@ -1174,14 +1177,14 @@ class ClaudeAdapter extends BaseAdapter { // never be reused. This check is deliberately independent of the // decision fingerprint: a failed decision fetch must not keep a // read-only plan process serving execute requests. - const modeStale = existingPP.spawnMode !== this._mode; + const modeStale = existingPP.spawnMode !== this._modeFor(msgChannel); const decisionsStale = decisionHash !== null && existingPP.decisionHash !== decisionHash; if (modeStale || decisionsStale) { // Kill it and fall through to a fresh spawn: --resume keeps the // conversation (it lives in the CLI transcript), while the new spawn // carries the current mode and re-pins the current decisions. this._log(modeStale - ? `Mode changed to ${this._mode} for ${msgChannel} — respawning with resume` + ? `Mode changed to ${this._modeFor(msgChannel)} for ${msgChannel} — respawning with resume` : `Decision log changed for ${msgChannel} — respawning with resume to re-pin decisions`); await this._killPersistentProc(msgChannel); } else { @@ -1304,7 +1307,7 @@ class ClaudeAdapter extends BaseAdapter { pp.decisionHash = decisionLogOpt ? decisionFingerprint(decisionLogOpt.entryId, decisionLogOpt.content) : decisionFingerprint(null, null); - pp.spawnMode = this._mode; + pp.spawnMode = this._modeFor(msgChannel); this._log(`Spawned persistent process for ${msgChannel} (attempt ${attempt + 1})`); const result = await this._sendToPersistentProc(pp, effectiveContent); diff --git a/packages/agent-connector/src/adapters/workspace-prompt.js b/packages/agent-connector/src/adapters/workspace-prompt.js index 1e9ad2d6a..c5014c114 100644 --- a/packages/agent-connector/src/adapters/workspace-prompt.js +++ b/packages/agent-connector/src/adapters/workspace-prompt.js @@ -598,6 +598,71 @@ function buildApiSkillsPrompt({ endpoint, workspaceId, token, agentName, channel return sections.join('\n'); } +/** + * Per-message directive for a channel whose requirement is still being + * clarified (the backend's phase gate, see `_apply_phase_gate` in + * workspace_mod.py). + * + * The gate already decides WHO gets woken; this is what makes the wake-up + * safe: an agent consulted mid-clarification answers the question instead of + * starting to build against a specification that isn't settled yet. + * + * `role` comes from the routed message's metadata: + * 'owner' → this agent holds the floor (phase owner / channel master) + * 'plan' → this agent was @mentioned but must not build (target_modes) + * other → phase is active but this agent is neither; just state the phase + * + * Returns '' when there is nothing to say, so callers can concatenate + * unconditionally. + */ +function buildPhaseGateDirective({ role, owner, endpoint, workspaceId, channelName } = {}) { + if (!role) return ''; + const who = owner || 'the phase owner'; + + if (role === 'owner') { + const base = endpoint ? String(endpoint).replace(/\/+$/, '') : ''; + const patch = base && workspaceId && channelName + ? ` (no such tool? PATCH ${base}/v1/workspaces/${workspaceId}/channels/${channelName} ` + + 'with {"phase":"building"} and your X-Workspace-Token header)' + : ''; + return ( + '\n\n---\n' + + '[Workspace phase: CLARIFYING — you own this phase]\n' + + 'The requirement in this channel is not settled yet, and settling it is ' + + 'your job. Ask what is still open, confirm your understanding, and record ' + + 'what the user agrees to. While this phase is active no other agent can ' + + 'start implementing — they can only be consulted.\n' + + 'Once the user has confirmed the requirement, advance the phase: call ' + + `\`workspace_set_phase\` with phase="building"${patch}. ` + + 'Nobody can start building until you do, so do not leave it behind — but ' + + 'do not advance it on your own judgement either; wait for the user.\n' + ); + } + + if (role === 'plan') { + return ( + '\n\n---\n' + + '[Workspace phase: CLARIFYING — answer in PLAN mode]\n' + + 'You have been pulled in while the requirement in this channel is still ' + + `being clarified by ${who}. Answer what was actually asked: feasibility, ` + + 'risks, options, rough effort, or questions of your own that need ' + + 'answering before this can be built.\n' + + 'Do NOT start the work — no code, no file edits, no commands that change ' + + 'anything. The specification is not final, so anything built now would be ' + + `built on guesses. ${who} advances the phase once the requirement is ` + + 'confirmed, and implementation starts then.\n' + ); + } + + return ( + '\n\n---\n' + + `[Workspace phase: CLARIFYING — owned by ${who}]\n` + + 'The requirement in this channel is still being clarified. Keep your reply ' + + 'to what helps settle it, and leave implementation until the phase ' + + 'advances.\n' + ); +} + /** * Guardrails shared across all adapter prompt builders. */ @@ -962,6 +1027,7 @@ module.exports = { buildBrowserDirective, buildCollaborationPrompt, buildModePrompt, + buildPhaseGateDirective, buildGuardrails, buildApiSkillsPrompt, buildClaudeMcpToolBlock, diff --git a/packages/agent-connector/src/mcp-server.js b/packages/agent-connector/src/mcp-server.js index d50b44e19..f7a21dbb7 100644 --- a/packages/agent-connector/src/mcp-server.js +++ b/packages/agent-connector/src/mcp-server.js @@ -52,6 +52,30 @@ function buildToolDefs(disabledModules) { required: ['status'], }, }, + { + name: 'workspace_set_phase', + description: + 'Set this channel\'s requirement phase. "clarifying" keeps the floor with the ' + + 'phase owner so no agent starts building on an unsettled spec (others can only be ' + + 'consulted, and only in plan mode); "building" releases the gate once the user has ' + + 'confirmed the requirement; "open" turns the gate off entirely. Only advance to ' + + '"building" after the user confirms — never on your own judgement.', + inputSchema: { + type: 'object', + properties: { + phase: { + type: 'string', + enum: ['open', 'clarifying', 'building'], + description: 'Phase to set for the current channel', + }, + owner: { + type: 'string', + description: 'Agent that owns the clarifying phase (defaults to the channel master)', + }, + }, + required: ['phase'], + }, + }, ]; // -- Files module -- @@ -636,6 +660,15 @@ class McpServer { return text(`Status updated: ${args.status}`); } + case 'workspace_set_phase': { + const result = await this.ws.setChannelPhase( + this.workspaceId, this.channelName, this.token, + { phase: args.phase, owner: args.owner }, + ); + const owner = result.phaseOwner ? ` (owner: ${result.phaseOwner})` : ''; + return text(`Channel phase set to "${result.phase || args.phase}"${owner}.`); + } + // ── Files ── case 'workspace_list_files': { diff --git a/packages/agent-connector/src/workspace-client.js b/packages/agent-connector/src/workspace-client.js index 772e36255..b637d01dc 100644 --- a/packages/agent-connector/src/workspace-client.js +++ b/packages/agent-connector/src/workspace-client.js @@ -389,12 +389,33 @@ class WorkspaceClient { titleManuallySet: result.titleManuallySet || false, resumeFrom: result.resumeFrom || null, status: result.status || 'active', + phase: result.phase || 'open', + phaseOwner: result.phaseOwner || null, }; } catch { - return { sessionId: channelName, title: channelName, status: 'active' }; + return { sessionId: channelName, title: channelName, status: 'active', phase: 'open', phaseOwner: null }; } } + /** + * Set a channel's clarification phase via PATCH + * /v1/workspaces/{id}/channels/{name}. + * + * Unlike updateSession this one throws: it backs an agent-facing tool, and + * silently failing to advance the phase would leave the thread gated with + * the agent believing it had opened it. + */ + async setChannelPhase(workspaceId, channelName, token, { phase, owner } = {}) { + const body = { phase }; + if (owner !== undefined) body.phase_owner = owner; + const data = await this._patch( + `/v1/workspaces/${workspaceId}/channels/${channelName}`, + body, + this._wsHeaders(token), + ); + return (data && (data.data || data)) || {}; + } + /** * Update session/channel info via PATCH /v1/workspaces/{id}/channels/{name}. */ diff --git a/packages/agent-connector/test/phase-gate.test.js b/packages/agent-connector/test/phase-gate.test.js new file mode 100644 index 000000000..22bf21807 --- /dev/null +++ b/packages/agent-connector/test/phase-gate.test.js @@ -0,0 +1,278 @@ +'use strict'; + +const { describe, it } = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); + +const BaseAdapter = require('../src/adapters/base'); +const ClaudeAdapter = require('../src/adapters/claude'); +const { buildPhaseGateDirective } = require('../src/adapters/workspace-prompt'); + +function makeAdapter(agentName = 'rd') { + return new BaseAdapter({ + workspaceId: 'ws-1', + channelName: 'session-1', + token: 't', + agentName, + endpoint: 'https://example.test', + }); +} + +function makeMsg(metadata, { content = 'build the sync module', channel = 'session-1' } = {}) { + return { + messageId: 'm-1', + sessionId: channel, + senderType: 'human', + senderName: 'user', + content, + mentions: [], + messageType: 'chat', + metadata, + }; +} + +describe('buildPhaseGateDirective', () => { + it('returns nothing without a role', () => { + assert.equal(buildPhaseGateDirective({}), ''); + assert.equal(buildPhaseGateDirective({ role: null }), ''); + }); + + it('forbids building in the plan role and names the owner', () => { + const text = buildPhaseGateDirective({ role: 'plan', owner: 'pm' }); + assert.match(text, /PLAN mode/); + assert.match(text, /Do NOT start the work/); + assert.match(text, /pm/); + }); + + it('tells the owner how to advance the phase', () => { + const text = buildPhaseGateDirective({ + role: 'owner', + owner: 'pm', + endpoint: 'https://example.test/', + workspaceId: 'ws-1', + channelName: 'session-1', + }); + assert.match(text, /workspace_set_phase/); + assert.match(text, /phase="building"/); + // Trailing slash on the endpoint must not double up in the fallback URL. + assert.match(text, /https:\/\/example\.test\/v1\/workspaces\/ws-1\/channels\/session-1/); + }); + + it('omits the REST fallback when the adapter has no endpoint context', () => { + const text = buildPhaseGateDirective({ role: 'owner', owner: 'pm' }); + assert.match(text, /workspace_set_phase/); + assert.doesNotMatch(text, /PATCH/); + }); + + it('states the phase for a member that is neither owner nor gated', () => { + const text = buildPhaseGateDirective({ role: 'member', owner: 'pm' }); + assert.match(text, /still being clarified/); + assert.doesNotMatch(text, /PLAN mode/); + }); +}); + +describe('BaseAdapter phase role', () => { + it('is null when the channel is not clarifying', () => { + const a = makeAdapter(); + assert.equal(a._phaseRole(makeMsg({})), null); + assert.equal(a._phaseRole(makeMsg({ phase: 'building', phase_owner: 'pm' })), null); + assert.equal(a._phaseRole({ content: 'x' }), null); + }); + + it('is plan when this agent was gated', () => { + const a = makeAdapter('rd'); + const msg = makeMsg({ phase: 'clarifying', phase_owner: 'pm', target_modes: { rd: 'plan' } }); + assert.equal(a._phaseRole(msg), 'plan'); + }); + + it('is owner for the phase owner', () => { + const a = makeAdapter('pm'); + const msg = makeMsg({ phase: 'clarifying', phase_owner: 'pm' }); + assert.equal(a._phaseRole(msg), 'owner'); + }); + + it('is member for another agent in a gated channel', () => { + const a = makeAdapter('qa'); + const msg = makeMsg({ phase: 'clarifying', phase_owner: 'pm', target_modes: { rd: 'plan' } }); + assert.equal(a._phaseRole(msg), 'member'); + }); + + it('does not read another agent\'s gate as its own', () => { + const a = makeAdapter('qa'); + const msg = makeMsg({ phase: 'clarifying', phase_owner: 'pm', target_modes: { rd: 'plan' } }); + assert.notEqual(a._phaseRole(msg), 'plan'); + }); +}); + +describe('BaseAdapter._applyPhaseGate', () => { + it('leaves an ungated message untouched', () => { + const a = makeAdapter(); + const msg = makeMsg({}); + assert.equal(a._applyPhaseGate(msg), msg); + }); + + it('appends the directive after the user content, never before', () => { + const a = makeAdapter('rd'); + const msg = makeMsg({ phase: 'clarifying', phase_owner: 'pm', target_modes: { rd: 'plan' } }); + const gated = a._applyPhaseGate(msg); + assert.ok(gated.content.startsWith('build the sync module')); + assert.match(gated.content, /Do NOT start the work/); + }); + + it('does not mutate the original message', () => { + const a = makeAdapter('rd'); + const msg = makeMsg({ phase: 'clarifying', phase_owner: 'pm', target_modes: { rd: 'plan' } }); + a._applyPhaseGate(msg); + assert.equal(msg.content, 'build the sync module'); + }); +}); + +describe('BaseAdapter mode override', () => { + it('defaults to the agent mode', () => { + const a = makeAdapter(); + assert.equal(a._modeFor('session-1'), 'execute'); + }); + + it('runs a gated message in plan mode and restores afterwards', async () => { + const a = makeAdapter('rd'); + const seen = []; + a._handleMessage = async (m) => { seen.push(a._modeFor(m.sessionId)); }; + + const gated = makeMsg({ phase: 'clarifying', phase_owner: 'pm', target_modes: { rd: 'plan' } }); + await a._runMessage('session-1', gated); + await a._runMessage('session-1', makeMsg({})); + + assert.deepEqual(seen, ['plan', 'execute']); + assert.equal(a._modeFor('session-1'), 'execute'); + }); + + it('clears the override even when the handler throws', async () => { + const a = makeAdapter('rd'); + a._handleMessage = async () => { throw new Error('boom'); }; + const gated = makeMsg({ phase: 'clarifying', phase_owner: 'pm', target_modes: { rd: 'plan' } }); + await assert.rejects(() => a._runMessage('session-1', gated), /boom/); + assert.equal(a._modeFor('session-1'), 'execute'); + }); + + it('keeps one channel\'s gate out of another channel', async () => { + const a = makeAdapter('rd'); + let release; + const started = new Promise((r) => { release = r; }); + a._handleMessage = async (m) => { + if (m.sessionId === 'session-1') { + release(); + await new Promise((r) => setTimeout(r, 20)); + } + }; + const gated = makeMsg( + { phase: 'clarifying', phase_owner: 'pm', target_modes: { rd: 'plan' } }, + { channel: 'session-1' }, + ); + const running = a._runMessage('session-1', gated); + await started; + // While session-1 is gated, an unrelated channel keeps executing. + assert.equal(a._modeFor('session-2'), 'execute'); + assert.equal(a._modeFor('session-1'), 'plan'); + await running; + }); +}); + +// --------------------------------------------------------------------------- +// The gate has to survive contact with the actual CLI invocation: a directive +// in the prompt is advice, the permission flags are enforcement. These assert +// on the argv the adapter would spawn, which is what the earlier tests (mode +// bookkeeping only) could not tell apart from a no-op. +// --------------------------------------------------------------------------- + +function claudeAdapter(workDir, toolMode) { + const adapter = new ClaudeAdapter({ + workspaceId: 'ws-1', + channelName: 'session-1', + token: 'tok', + agentName: 'rd', + workingDir: workDir, + toolMode, + }); + adapter._log = () => {}; + return adapter; +} + +/** + * Run `fn` with a throwaway working directory AND a throwaway home. + * + * The MCP command builder writes its generated config under + * `os.homedir()/.openagents/mcp-configs`, so isolating only the working + * directory would leak files into the developer's real home and fail + * outright where HOME is read-only (sandboxed CI, containers). + */ +function withWorkDir(fn) { + const workDir = fs.mkdtempSync(path.join(os.tmpdir(), 'oa-phase-gate-')); + const realHomedir = os.homedir; + os.homedir = () => workDir; + try { + return fn(workDir); + } finally { + os.homedir = realHomedir; + fs.rmSync(workDir, { recursive: true, force: true }); + } +} + +describe('ClaudeAdapter honours the phase-gate plan override', () => { + it('skills mode: gated channel gets plan permissions and no write tools', () => { + withWorkDir((workDir) => { + const adapter = claudeAdapter(workDir, 'skills'); + adapter._modeOverrides['session-1'] = 'plan'; + + const cmd = []; + adapter._buildSkillsCmd(cmd, 'session-1'); + + assert.ok(cmd.includes('--permission-mode'), 'must pass --permission-mode'); + assert.equal(cmd[cmd.indexOf('--permission-mode') + 1], 'plan'); + assert.ok(!cmd.includes('--dangerously-skip-permissions')); + assert.ok(!cmd.includes('Write')); + assert.ok(!cmd.includes('Edit')); + }); + }); + + it('skills mode: an ungated channel still runs with write tools', () => { + withWorkDir((workDir) => { + const adapter = claudeAdapter(workDir, 'skills'); + const cmd = []; + adapter._buildSkillsCmd(cmd, 'session-1'); + + assert.ok(cmd.includes('--dangerously-skip-permissions')); + assert.ok(cmd.includes('Write')); + assert.ok(!cmd.includes('--permission-mode')); + }); + }); + + it('mcp mode: gated channel gets plan permissions and no write MCP tools', () => { + withWorkDir((workDir) => { + const adapter = claudeAdapter(workDir, 'mcp'); + adapter._modeOverrides['session-1'] = 'plan'; + + const cmd = []; + adapter._buildMcpCmd(cmd, 'session-1'); + + assert.equal(cmd[cmd.indexOf('--permission-mode') + 1], 'plan'); + assert.ok(!cmd.includes('--dangerously-skip-permissions')); + assert.ok(!cmd.includes('Write')); + assert.ok(!cmd.some((a) => a.endsWith('workspace_write_file'))); + }); + }); + + it('the override is scoped to the gated channel only', () => { + withWorkDir((workDir) => { + const adapter = claudeAdapter(workDir, 'skills'); + adapter._modeOverrides['session-1'] = 'plan'; + + const other = []; + adapter._buildSkillsCmd(other, 'session-2'); + + assert.ok(other.includes('--dangerously-skip-permissions')); + assert.ok(other.includes('Write')); + }); + }); +}); diff --git a/sdk/src/openagents/adapters/base.py b/sdk/src/openagents/adapters/base.py index e3a92562b..e7e93b6dc 100644 --- a/sdk/src/openagents/adapters/base.py +++ b/sdk/src/openagents/adapters/base.py @@ -21,6 +21,7 @@ from openagents.workspace_client import WorkspaceClient, DEFAULT_ENDPOINT from openagents.adapters.utils import generate_session_title, SESSION_DEFAULT_RE +from openagents.adapters.workspace_prompt import build_phase_gate_directive logger = logging.getLogger(__name__) @@ -47,6 +48,11 @@ def __init__( self._processed_ids: set = set() self._titled_sessions: set = set() self._mode: str = "execute" + # Per-channel, per-message mode override set by the clarification + # phase gate (see _mode_for). Keyed by channel — the same channel is + # processed serially, but different channels run in parallel, so a + # single scalar would leak one channel's gate onto another. + self._mode_overrides: dict[str, str] = {} self._last_control_id: Optional[str] = None self._control_wake_event = asyncio.Event() # Per-channel task tracking for parallel execution @@ -233,8 +239,75 @@ def _is_channel_busy(self, channel: str) -> bool: task = self._channel_tasks.get(channel) return task is not None and not task.done() + def _phase_role(self, msg: dict) -> Optional[str]: + """Role this agent plays in an active clarification phase. + + Returns ``"plan"`` when the backend downgraded this wake-up (answer, + don't build), ``"owner"`` when this agent owns the phase and is the + one who advances it, ``"member"`` when the phase is active but this + agent is neither, and None when the channel isn't gated. + """ + meta = (msg or {}).get("metadata") or {} + if meta.get("phase") != "clarifying": + return None + modes = meta.get("target_modes") or {} + if modes.get(self.agent_name) == "plan": + return "plan" + if meta.get("phase_owner") and meta.get("phase_owner") == self.agent_name: + return "owner" + return "member" + + def _apply_phase_gate(self, msg: dict) -> dict: + """Append the clarification-phase directive to a routed message. + + The backend gate decides who may be woken; this is what stops a woken + builder from implementing against an unsettled spec. Appended (not + prepended) so channel auto-titling still sees the user's own words + first. + """ + role = self._phase_role(msg) + if not role: + return msg + meta = msg.get("metadata") or {} + directive = build_phase_gate_directive( + role, + owner=meta.get("phase_owner"), + endpoint=self.endpoint, + workspace_id=self.workspace_id, + channel_name=msg.get("sessionId") or self.channel_name, + ) + if not directive: + return msg + logger.info(f"Phase gate active ({role}) for message {msg.get('messageId') or '?'}") + gated = dict(msg) + gated["content"] = f"{msg.get('content') or ''}{directive}" + return gated + + def _mode_for(self, channel: str) -> str: + """The mode this agent must run in for work on ``channel`` right now. + + The agent's own mode, unless the message being handled was gated into + PLAN by the clarification phase. Adapters that enforce plan mode at + the runtime level should read this instead of ``self._mode``. + """ + return self._mode_overrides.get(channel) or self._mode + + async def _run_message(self, channel: str, msg: dict): + """Run one message with its phase-gate mode override in effect.""" + if self._phase_role(msg) == "plan": + self._mode_overrides[channel] = "plan" + else: + self._mode_overrides.pop(channel, None) + try: + await self._handle_message(msg) + finally: + self._mode_overrides.pop(channel, None) + async def _dispatch_message(self, msg: dict): """Route a message to its channel — run in parallel or queue if busy.""" + # Carry the phase directive on the message itself so a queued message + # still holds the constraint it arrived under when it is finally run. + msg = self._apply_phase_gate(msg) channel = msg.get("sessionId") or self.channel_name if self._is_channel_busy(channel): @@ -261,7 +334,7 @@ async def _dispatch_message(self, msg: dict): async def _channel_worker(self, channel: str, msg: dict): """Process a message and then drain the channel's queue.""" try: - await self._handle_message(msg) + await self._run_message(channel, msg) except Exception as e: logger.exception(f"Error in channel worker for {channel}: {e}") try: @@ -275,7 +348,7 @@ async def _channel_worker(self, channel: str, msg: dict): break next_msg = queue.pop(0) try: - await self._handle_message(next_msg) + await self._run_message(channel, next_msg) except Exception as e: logger.exception(f"Error processing queued message in {channel}: {e}") try: diff --git a/sdk/src/openagents/adapters/claude.py b/sdk/src/openagents/adapters/claude.py index 06589d905..0d455bc09 100644 --- a/sdk/src/openagents/adapters/claude.py +++ b/sdk/src/openagents/adapters/claude.py @@ -242,7 +242,7 @@ def _build_claude_cmd(self, prompt: str, channel_name: str, browser_enabled: boo agent_name=self.agent_name, workspace_id=self.workspace_id, channel_name=channel_name, - mode=self._mode, + mode=self._mode_for(channel_name), browser_enabled=browser_enabled, ) @@ -287,7 +287,11 @@ def _build_claude_cmd(self, prompt: str, channel_name: str, browser_enabled: boo mcp_tools.append(f"{_pfx}tunnel_list") mcp_write_tools += [f"{_pfx}tunnel_expose", f"{_pfx}tunnel_close"] - if self._mode == "plan": + # `_mode_for`, not `_mode`: a message the clarification gate + # downgraded has to launch with plan permissions and read-only tools, + # otherwise the "don't build yet" constraint is only prompt text the + # model can talk itself out of. + if self._mode_for(channel_name) == "plan": cmd.extend(["--permission-mode", "plan"]) allowed = mcp_tools + ["Read", "Glob", "Grep"] else: diff --git a/sdk/src/openagents/adapters/workspace_prompt.py b/sdk/src/openagents/adapters/workspace_prompt.py index 55def7562..7599c453c 100644 --- a/sdk/src/openagents/adapters/workspace_prompt.py +++ b/sdk/src/openagents/adapters/workspace_prompt.py @@ -104,6 +104,77 @@ def build_collaboration_prompt() -> str: ) +def build_phase_gate_directive( + role: Optional[str], + owner: Optional[str] = None, + endpoint: Optional[str] = None, + workspace_id: Optional[str] = None, + channel_name: Optional[str] = None, +) -> str: + """Per-message directive for a channel that is still clarifying. + + The backend's phase gate (``_apply_phase_gate`` in workspace_mod.py) + decides who may be woken; this is what makes the wake-up safe — an agent + consulted mid-clarification answers the question instead of building + against a specification that isn't settled. + + ``role`` comes from the routed message's metadata: ``"owner"`` (holds the + floor), ``"plan"`` (mentioned, must not build), anything else (phase is + active but this agent is neither). Returns "" when there is nothing to + say, so callers can concatenate unconditionally. + + Mirrors ``buildPhaseGateDirective`` in + packages/agent-connector/src/adapters/workspace-prompt.js. + """ + if not role: + return "" + who = owner or "the phase owner" + + if role == "owner": + patch = "" + if endpoint and workspace_id and channel_name: + base = endpoint.rstrip("/") + patch = ( + f' (no such tool? PATCH {base}/v1/workspaces/{workspace_id}' + f'/channels/{channel_name} with {{"phase":"building"}} and your ' + "X-Workspace-Token header)" + ) + return ( + "\n\n---\n" + "[Workspace phase: CLARIFYING — you own this phase]\n" + "The requirement in this channel is not settled yet, and settling it " + "is your job. Ask what is still open, confirm your understanding, and " + "record what the user agrees to. While this phase is active no other " + "agent can start implementing — they can only be consulted.\n" + "Once the user has confirmed the requirement, advance the phase: call " + f'`workspace_set_phase` with phase="building"{patch}. ' + "Nobody can start building until you do, so do not leave it behind — " + "but do not advance it on your own judgement either; wait for the user.\n" + ) + + if role == "plan": + return ( + "\n\n---\n" + "[Workspace phase: CLARIFYING — answer in PLAN mode]\n" + "You have been pulled in while the requirement in this channel is " + f"still being clarified by {who}. Answer what was actually asked: " + "feasibility, risks, options, rough effort, or questions of your own " + "that need answering before this can be built.\n" + "Do NOT start the work — no code, no file edits, no commands that " + "change anything. The specification is not final, so anything built " + f"now would be built on guesses. {who} advances the phase once the " + "requirement is confirmed, and implementation starts then.\n" + ) + + return ( + "\n\n---\n" + f"[Workspace phase: CLARIFYING — owned by {who}]\n" + "The requirement in this channel is still being clarified. Keep your " + "reply to what helps settle it, and leave implementation until the " + "phase advances.\n" + ) + + def build_mode_prompt(mode: str) -> str: """Build mode-specific instructions.""" if mode == "plan": diff --git a/tests/test_workspace_phase_gate.py b/tests/test_workspace_phase_gate.py new file mode 100644 index 000000000..65c2bee59 --- /dev/null +++ b/tests/test_workspace_phase_gate.py @@ -0,0 +1,206 @@ +""" +Tests for the clarification phase gate in the Python adapter stack. + +The backend decides who may be woken while a channel's requirement is still +being clarified; these cover the adapter half — the directive an agent is +given and the per-message PLAN downgrade that stops a consulted builder from +starting work. Mirrors packages/agent-connector/test/phase-gate.test.js. +""" +import asyncio + +import pytest + +from openagents.adapters.base import BaseAdapter +from openagents.adapters.workspace_prompt import build_phase_gate_directive + + +class _Adapter(BaseAdapter): + """Concrete adapter that records the mode each message ran under.""" + + def __init__(self, agent_name="rd"): + super().__init__( + workspace_id="ws-1", + channel_name="session-1", + token="t", + agent_name=agent_name, + endpoint="https://example.test", + ) + self.seen_modes = [] + self.raise_on_handle = False + + async def _handle_message(self, msg: dict): + if self.raise_on_handle: + raise RuntimeError("boom") + self.seen_modes.append(self._mode_for(msg.get("sessionId") or self.channel_name)) + + +def _msg(metadata, content="build the sync module", channel="session-1"): + return { + "messageId": "m-1", + "sessionId": channel, + "senderType": "human", + "senderName": "user", + "content": content, + "metadata": metadata, + } + + +CLARIFYING = {"phase": "clarifying", "phase_owner": "pm", "target_modes": {"rd": "plan"}} + + +class TestPhaseGateDirective: + def test_no_role_emits_nothing(self): + assert build_phase_gate_directive(None) == "" + assert build_phase_gate_directive("") == "" + + def test_plan_role_forbids_building(self): + out = build_phase_gate_directive("plan", owner="pm") + assert "PLAN mode" in out + assert "Do NOT start the work" in out + assert "pm" in out + + def test_owner_role_explains_how_to_advance(self): + out = build_phase_gate_directive( + "owner", owner="pm", + endpoint="https://example.test/", + workspace_id="ws-1", + channel_name="session-1", + ) + assert "workspace_set_phase" in out + assert 'phase="building"' in out + # A trailing slash on the endpoint must not double up in the URL. + assert "https://example.test/v1/workspaces/ws-1/channels/session-1" in out + + def test_owner_role_without_endpoint_omits_rest_fallback(self): + out = build_phase_gate_directive("owner", owner="pm") + assert "workspace_set_phase" in out + assert "PATCH" not in out + + def test_member_role_states_the_phase_only(self): + out = build_phase_gate_directive("member", owner="pm") + assert "still being clarified" in out + assert "PLAN mode" not in out + + +class TestPhaseRole: + def test_none_when_not_clarifying(self): + a = _Adapter() + assert a._phase_role(_msg({})) is None + assert a._phase_role(_msg({"phase": "building", "phase_owner": "pm"})) is None + + def test_plan_for_the_gated_agent(self): + assert _Adapter("rd")._phase_role(_msg(CLARIFYING)) == "plan" + + def test_owner_for_the_phase_owner(self): + a = _Adapter("pm") + assert a._phase_role(_msg({"phase": "clarifying", "phase_owner": "pm"})) == "owner" + + def test_another_agents_gate_is_not_read_as_own(self): + assert _Adapter("qa")._phase_role(_msg(CLARIFYING)) == "member" + + +class TestApplyPhaseGate: + def test_ungated_message_untouched(self): + a = _Adapter() + msg = _msg({}) + assert a._apply_phase_gate(msg) is msg + + def test_directive_is_appended_not_prepended(self): + gated = _Adapter("rd")._apply_phase_gate(_msg(CLARIFYING)) + assert gated["content"].startswith("build the sync module") + assert "Do NOT start the work" in gated["content"] + + def test_original_message_not_mutated(self): + msg = _msg(CLARIFYING) + _Adapter("rd")._apply_phase_gate(msg) + assert msg["content"] == "build the sync module" + + +class TestModeOverride: + def test_defaults_to_agent_mode(self): + assert _Adapter()._mode_for("session-1") == "execute" + + def test_gated_message_runs_in_plan_then_restores(self): + a = _Adapter("rd") + asyncio.run(a._run_message("session-1", _msg(CLARIFYING))) + asyncio.run(a._run_message("session-1", _msg({}))) + assert a.seen_modes == ["plan", "execute"] + assert a._mode_for("session-1") == "execute" + + def test_override_cleared_when_handler_raises(self): + a = _Adapter("rd") + a.raise_on_handle = True + with pytest.raises(RuntimeError): + asyncio.run(a._run_message("session-1", _msg(CLARIFYING))) + assert a._mode_for("session-1") == "execute" + + def test_one_channels_gate_does_not_leak_into_another(self): + a = _Adapter("rd") + asyncio.run(a._run_message("session-1", _msg(CLARIFYING))) + # session-2 never carried a gate. + assert a._mode_for("session-2") == "execute" + + +class TestClaudeCommandEnforcement: + """The directive is advice; the CLI flags are enforcement. These assert on + the argv the adapter would actually spawn — mode bookkeeping alone cannot + tell a working gate apart from a no-op.""" + + def _adapter(self, tmp_path, monkeypatch): + from openagents.adapters.claude import ClaudeAdapter + + # Contain the adapter's home-directory writes (session store, MCP + # config) inside the test's tmp dir. + monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path) + monkeypatch.setattr( + "openagents.adapters.claude.shutil.which", + lambda name: f"/usr/bin/{name}", + ) + return ClaudeAdapter( + workspace_id="ws-1", + channel_name="session-1", + token="tok", + agent_name="rd", + working_dir=str(tmp_path), + ) + + def test_gated_channel_gets_plan_permissions(self, tmp_path, monkeypatch): + adapter = self._adapter(tmp_path, monkeypatch) + adapter._mode_overrides["session-1"] = "plan" + + cmd = adapter._build_claude_cmd("do the thing", "session-1") + + assert "--permission-mode" in cmd + assert cmd[cmd.index("--permission-mode") + 1] == "plan" + assert "--dangerously-skip-permissions" not in cmd + assert "Write" not in cmd + assert "Edit" not in cmd + assert not any(a.endswith("workspace_write_file") for a in cmd) + + def test_ungated_channel_keeps_write_tools(self, tmp_path, monkeypatch): + adapter = self._adapter(tmp_path, monkeypatch) + + cmd = adapter._build_claude_cmd("do the thing", "session-1") + + assert "--dangerously-skip-permissions" in cmd + assert "Write" in cmd + assert "--permission-mode" not in cmd + + def test_override_is_scoped_to_the_gated_channel(self, tmp_path, monkeypatch): + adapter = self._adapter(tmp_path, monkeypatch) + adapter._mode_overrides["session-1"] = "plan" + + cmd = adapter._build_claude_cmd("do the thing", "session-2") + + assert "--dangerously-skip-permissions" in cmd + assert "Write" in cmd + + def test_gated_channel_gets_plan_system_prompt(self, tmp_path, monkeypatch): + adapter = self._adapter(tmp_path, monkeypatch) + adapter._mode_overrides["session-1"] = "plan" + + cmd = adapter._build_claude_cmd("do the thing", "session-1") + system_prompt = cmd[cmd.index("--append-system-prompt") + 1] + + assert "You are in PLAN mode" in system_prompt + assert "Do not make edits" in system_prompt diff --git a/workspace/backend/alembic/versions/029_add_channel_phase.py b/workspace/backend/alembic/versions/029_add_channel_phase.py new file mode 100644 index 000000000..c85b74c7c --- /dev/null +++ b/workspace/backend/alembic/versions/029_add_channel_phase.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +"""Add the requirement-clarification phase gate to channels. + +Two columns backing a per-thread gate that stops execution agents from +starting work while the requirement is still being clarified: + + phase "open" (no gate, current behaviour) | "clarifying" | "building" + phase_owner agent that owns the clarifying phase; falls back to master_agent + +`phase` is NOT NULL with a server default of 'open', so every existing channel +keeps today's routing untouched until a thread is explicitly gated. + +Revision ID: 029 +Revises: 028 +Create Date: 2026-08-02 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "029" +down_revision = "028" +branch_labels = None +depends_on = None + + +def _has_column(inspector, table, column): + if table not in inspector.get_table_names(): + return False + return any(c["name"] == column for c in inspector.get_columns(table)) + + +def upgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + + if not _has_column(inspector, "channels", "phase"): + op.add_column( + "channels", + sa.Column( + "phase", + sa.Text(), + server_default=sa.text("'open'"), + nullable=False, + ), + ) + if not _has_column(inspector, "channels", "phase_owner"): + op.add_column("channels", sa.Column("phase_owner", sa.Text(), nullable=True)) + + +def downgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + + for column in ("phase_owner", "phase"): + if _has_column(inspector, "channels", column): + op.drop_column("channels", column) diff --git a/workspace/backend/app/models.py b/workspace/backend/app/models.py index 0123704a8..5261b14c0 100644 --- a/workspace/backend/app/models.py +++ b/workspace/backend/app/models.py @@ -151,6 +151,20 @@ class Channel(Base): # Free-text collaboration plan (with @agent mentions) used only in # "workflow" mode; injected into the router prompt as the routing policy. orchestration_instruction = Column(Text, nullable=True) + # Requirement-clarification gate, orthogonal to orchestration_mode: + # "open" → no gate; routing behaves exactly as it always has [default] + # "clarifying" → the requirement is still being clarified. Routing is + # pinned to the phase owner (or the master); any other + # agent can only be woken by an explicit @mention, and + # then only in PLAN mode — so a builder agent can be + # consulted but cannot start implementing. + # "building" → clarification finished; routing is unrestricted again. + # Default is "open" so existing threads keep their current behaviour until + # someone opts a thread into the gate. + phase = Column(Text, nullable=False, server_default=text("'open'")) + # Agent that owns the clarifying phase (typically the PM/requirements + # agent). Falls back to `master_agent` when unset. + phase_owner = Column(Text, nullable=True) status = Column(Text, default="active") # active | archived | deleted starred = Column(Boolean, default=False, server_default=text("FALSE")) last_event_at = Column(BigInteger, nullable=True) diff --git a/workspace/backend/app/mods/workspace_mod.py b/workspace/backend/app/mods/workspace_mod.py index dc1de3d41..61e1fe5f9 100644 --- a/workspace/backend/app/mods/workspace_mod.py +++ b/workspace/backend/app/mods/workspace_mod.py @@ -16,7 +16,7 @@ import logging import re from datetime import datetime, timezone -from typing import List, Optional +from typing import Dict, List, Optional, Tuple from sqlalchemy import select @@ -223,12 +223,17 @@ async def _handle_agent_remove(event: Event, ctx: PipelineContext) -> Optional[E # If removed agent was master, promote the next available (non-removed) agent if was_master: + # `.scalars().first()`, not `.scalar_one_or_none()`: the query is a + # "pick the longest-serving survivor" over every remaining member, so + # two or more survivors made it raise MultipleResultsFound and the + # whole removal failed — which also aborted the phase-owner repair + # below. next_master = db.execute( select(WorkspaceMember).where( WorkspaceMember.workspace_id == workspace.id, WorkspaceMember.status != "removed", ).order_by(WorkspaceMember.joined_at.asc()) - ).scalar_one_or_none() + ).scalars().first() if next_master: next_master.role = "master" @@ -247,6 +252,19 @@ async def _handle_agent_remove(event: Event, ctx: PipelineContext) -> Optional[E ch.master_agent = new_master_name db.flush() + # Repair any clarification gate this agent was holding. Left alone, the + # gate would keep redirecting to a removed agent and every message in + # that thread would be routed to nobody. + owned = db.execute( + select(Channel).where( + Channel.workspace_id == workspace.id, + Channel.phase_owner == agent_name, + ) + ).scalars().all() + for ch in owned: + _reassign_phase_owner(ch, db, workspace, leaving=agent_name) + db.flush() + event.metadata["removed_agent"] = agent_name if new_master_name: event.metadata["new_master"] = new_master_name @@ -309,12 +327,67 @@ async def _handle_ping(event: Event, ctx: PipelineContext) -> Optional[Event]: async def _handle_channel_create(event: Event, ctx: PipelineContext) -> Optional[Event]: """network.channel.create → create Channel + initial ChannelMember rows.""" - from app.models import Channel, ChannelMember, ChannelHumanMember, WorkspaceCollaborator + from app.models import ( + Channel, ChannelMember, ChannelHumanMember, WorkspaceCollaborator, + WorkspaceMember, + ) db = ctx.extra["db"] workspace = ctx.extra["workspace"] payload = event.payload or {} + # A thread can start gated (the creator already knows the requirement + # needs clarifying) instead of being PATCHed a moment after creation, + # which would leave a window where a builder could be woken. + phase = (payload.get("phase") or PHASE_OPEN).strip().lower() + if phase not in CHANNEL_PHASES: + phase = PHASE_OPEN + phase_owner = (payload.get("phase_owner") or "").strip() or None + if phase == PHASE_CLARIFYING: + # A gate needs someone to hold the floor. The owner must be among the + # agents this channel is being created with, otherwise the channel + # would be born in the one state the gate cannot enforce — showing + # "clarifying" in the UI while routing behaves as if it were open. + initial = [ + a for a in (payload.get("participants") or []) + if a and a != "__no_response__" + ] + resolved = phase_owner or payload.get("master") + # The participants list is caller-supplied and unverified, so being + # named in it proves nothing — the owner must also be a real, live + # member of this workspace, exactly as PATCH requires. + known = False + if resolved: + owner_member = db.execute( + select(WorkspaceMember).where( + WorkspaceMember.workspace_id == workspace.id, + WorkspaceMember.agent_name == resolved, + ) + ).scalar_one_or_none() + known = bool(owner_member) and (owner_member.status or "").lower() != "removed" + if not resolved or resolved not in initial or not known: + # Do NOT quietly create this thread open. The caller explicitly + # asked for a gate, and the owner can go stale between the client + # listing agents and this handler running (removed, taken + # offline). Creating it open would leave the client believing the + # thread is protected while its very first message routes to a + # builder — the failure mode this whole feature exists to remove. + # + # Keep the gate and leave it unowned instead: routing degrades to + # "everyone answers in plan mode" (see `_apply_phase_gate`), so + # nothing can be built, and the thread renders as "Clarifying · + # needs an owner" until a human names one. Rejecting the event + # outright would throw away the whole thread the user just set up. + logger.warning( + "workspace_mod: channel.create asked for phase=clarifying with " + "owner %r not among live participants %s — creating it gated " + "but unowned (plan-only) so nothing starts building", + resolved, initial, + ) + phase_owner = None + else: + phase_owner = resolved + channel = Channel( workspace_id=workspace.id, name=payload.get("name", f"channel-{event.id[:8]}"), @@ -322,6 +395,8 @@ async def _handle_channel_create(event: Event, ctx: PipelineContext) -> Optional created_by=event.source, master_agent=payload.get("master"), resume_from=payload.get("resume_from"), + phase=phase, + phase_owner=phase_owner, status="active", ) db.add(channel) @@ -372,6 +447,12 @@ async def _handle_channel_create(event: Event, ctx: PipelineContext) -> Optional # Enrich event with created channel info event.metadata["channel_id"] = str(channel.id) event.metadata["channel_name"] = channel.name + # What was actually persisted, which is not always what was asked for + # (an owner can go stale between the client reading the agent list and + # this handler running). Clients must render from these, never from the + # values they sent. + event.metadata["phase"] = channel.phase + event.metadata["phase_owner"] = channel.phase_owner event.target = f"channel/{channel.name}" return event @@ -505,6 +586,27 @@ async def _handle_channel_leave(event: Event, ctx: PipelineContext) -> Optional[ if member: db.delete(member) db.flush() + db.refresh(channel) + # An agent that left the channel must stop being its master. Both + # `_master_targets` and `_fallback_targets` route to `master_agent` + # unconditionally, so leaving the field pointing at a departed agent + # sends every later message to someone who is no longer listening. + # Clearing it hands routing to the online-participant fallback. + # (Pre-existing on this path; it also decides where the phase gate + # below can hand the floor, so the two are repaired together.) + if channel.master_agent == agent_name: + channel.master_agent = None + db.flush() + logger.info( + "workspace_mod: %s left %s and was its master — master cleared", + agent_name, channel.name, + ) + # The agent that just left may have been holding the clarification + # gate for this channel; hand it on (or open the thread) so routing + # never points at a non-participant. + if (getattr(channel, "phase_owner", None) or None) == agent_name: + _reassign_phase_owner(channel, db, workspace, leaving=agent_name) + db.flush() return event @@ -638,6 +740,226 @@ def _master_targets(event, channel, mentions: List[str]) -> List[str]: return [master] +# --------------------------------------------------------------------------- +# Requirement-clarification phase gate +# +# Without it, routing is a stateless per-message decision that is biased +# towards "somebody must answer now" — so a builder agent gets handed a +# half-specified request and starts implementing while the requirement is +# still being clarified. The gate makes "we are still clarifying" a state +# the router has to obey rather than a convention agents are asked to honour. +# --------------------------------------------------------------------------- + +PHASE_OPEN = "open" +PHASE_CLARIFYING = "clarifying" +PHASE_BUILDING = "building" +CHANNEL_PHASES = (PHASE_OPEN, PHASE_CLARIFYING, PHASE_BUILDING) + + +def _channel_phase(channel) -> str: + """Current phase of a channel, defaulting to the ungated 'open'.""" + return (getattr(channel, "phase", None) or PHASE_OPEN).lower() + + +def _valid_gatekeeper_names(channel, db, workspace, candidates: List[str]) -> List[str]: + """Filter candidate gatekeepers down to ones that can actually take the + floor right now: a participant of THIS channel that is live by the same + `_member_is_online` rule the rest of routing uses. Order is preserved. + + Liveness, not just membership. A crashed daemon leaves `status='online'` + in the database indefinitely, so a "still a member" check would keep + handing the floor to an agent with no connector behind it: the thread + looks gated and healthy while every message goes unanswered. Membership + also changes after the phase is set (agent removed, agent left the + channel), which write-time validation cannot cover. + + Dropping an offline owner from the gatekeepers does not open the gate — + an unowned gate holds everyone in plan mode (see `_apply_phase_gate`), so + the thread keeps answering without anyone building, and ownership resumes + by itself once the owner's daemon heartbeats again. `phase_owner` is + deliberately left untouched in the database; it records who owns the + requirement, not who happens to be connected this second. + """ + from app.models import WorkspaceMember + + if not candidates: + return [] + participants = {p.agent_name for p in (channel.participants or [])} + rows = db.execute( + select(WorkspaceMember).where( + WorkspaceMember.workspace_id == workspace.id, + WorkspaceMember.agent_name.in_(candidates), + ) + ).scalars().all() + live = {m.agent_name for m in rows if _member_is_online(m)} + return [n for n in candidates if n in live and n in participants] + + +def _phase_gatekeepers(channel, db, workspace) -> List[str]: + """Agents that keep the floor while the channel is clarifying. + + The phase owner (the requirements/PM agent) comes first and is the + redirect target; the channel master is included too, so a thread whose + owner is unset — or whose master coordinates alongside the owner — still + has somewhere to route. Order matters: callers redirect to the first + entry. + + Both are validated against live membership (see + `_valid_gatekeeper_names`). Returns [] when none survive — the gate is + then UNOWNED, not off: `_apply_phase_gate` keeps whoever was targeted but + holds them all in plan mode, so the thread answers and nobody builds. It + never falls back to unrestricted routing; only a human takes a gate off. + """ + owner = getattr(channel, "phase_owner", None) + names = [n for n in (owner, channel.master_agent) if n] + seen, ordered = set(), [] + for n in names: + if n not in seen: + seen.add(n) + ordered.append(n) + return _valid_gatekeeper_names(channel, db, workspace, ordered) + + +def _reassign_phase_owner(channel, db, workspace, leaving: Optional[str] = None) -> Optional[str]: + """Hand a channel's clarification gate to someone who can still hold it. + + Falls back to the channel master when that is a valid gatekeeper. + Otherwise the owner is cleared but THE GATE STAYS ON: routing then holds + every agent in plan mode (see `_apply_phase_gate`), so the thread keeps + answering without anyone building, and the UI shows it needs an owner. + Opening the gate here instead would silently undo a constraint the user + asked for — the same silent-ungating this feature exists to prevent — + just because an agent left. Only a human removes a gate. + + Picking an arbitrary surviving participant is deliberately not done: + ownership of the requirement is a human's call. + + Returns the new owner, or None when the gate was left unowned. + """ + master = channel.master_agent + candidates = [n for n in (master,) if n and n != leaving] + valid = _valid_gatekeeper_names(channel, db, workspace, candidates) + if valid: + channel.phase_owner = valid[0] + logger.info( + "phase gate: channel %s owner %s is gone — handed to %s", + channel.name, leaving, valid[0], + ) + return valid[0] + + channel.phase_owner = None + if _channel_phase(channel) == PHASE_CLARIFYING: + logger.warning( + "phase gate: channel %s lost its owner %s and has no valid master — " + "gate left on but unowned (plan-only) until a human names one", + channel.name, leaving, + ) + return None + + +def _apply_phase_gate( + event: Event, channel, targets: List[str], mentions: List[str], db, workspace, +) -> Tuple[List[str], Dict[str, str]]: + """Constrain routing while the channel's requirement is being clarified. + + Returns ``(targets, target_modes)``. ``target_modes`` maps an agent name + to the mode it must run in for THIS message; the connector downgrades a + ``"plan"`` target so it proposes and asks instead of building. + + Rules while ``phase == 'clarifying'`` (no-op in any other phase): + + • a gatekeeper target (owner / master) is passed through untouched — + clarification is their job; + • a non-gatekeeper that the sender explicitly @mentioned is kept, but + forced into PLAN mode. Naming an agent is a real request ("@rd is + this feasible?") and answering it is useful; starting to build off + an unconfirmed spec is not; + • any other non-gatekeeper target is dropped and the turn is handed to + the phase owner. This is the case that produces the reported bug — + the router picking the builder purely on topic match. + + When every target is dropped and the owner is the one who just spoke, the + turn ends (empty list → the caller's ``__no_response__`` sentinel) rather + than looping the owner back onto itself. + """ + if _channel_phase(channel) != PHASE_CLARIFYING: + return targets, {} + + gatekeepers = _phase_gatekeepers(channel, db, workspace) + if not gatekeepers: + # Nobody can hold the floor (owner removed mid-thread, legacy row). + # Neither extreme is acceptable here: passing routing through + # unchanged wakes the builder in execute mode — the exact behaviour + # this feature exists to prevent — and dropping every target strands + # the human with no reply. So keep whoever was targeted, but let + # nobody build: the thread stays responsive while the requirement is + # still officially unsettled. + logger.warning( + "phase gate: channel %s is clarifying but no gatekeeper can take the " + "floor (owner=%r, master=%r) — keeping targets %s in plan mode", + channel.name, getattr(channel, "phase_owner", None), + channel.master_agent, targets, + ) + return targets, {t: "plan" for t in targets} + + source = event.source or "" + sender = source[len("openagents:"):] if source.startswith("openagents:") else None + mention_set = set(mentions or []) + + kept: List[str] = [] + modes: Dict[str, str] = {} + dropped: List[str] = [] + for name in targets: + if name in gatekeepers: + kept.append(name) + elif name in mention_set: + kept.append(name) + modes[name] = "plan" + else: + dropped.append(name) + + if not kept: + # Hand the turn back to the owner — unless the owner is the sender, + # in which case there is nothing to hand back and the thread waits + # for the human. + redirect = next((g for g in gatekeepers if g != sender), None) + if redirect: + kept = [redirect] + + if dropped: + logger.info( + "phase gate: channel %s clarifying — dropped %s, routing to %s", + channel.name, dropped, kept or ["(nobody)"], + ) + + return kept, {k: v for k, v in modes.items() if k in kept} + + +def _phase_router_block(channel, db, workspace) -> str: + """Router-prompt block describing an active clarification phase. + + The gate is authoritative either way, but telling the router about the + phase keeps its decisions (and the logged reasoning) consistent with what + the gate will allow, instead of having every turn overridden after the + fact. + """ + if _channel_phase(channel) != PHASE_CLARIFYING: + return "" + gatekeepers = _phase_gatekeepers(channel, db, workspace) + if not gatekeepers: + return "" + owner = gatekeepers[0] + return ( + "\nCURRENT PHASE: CLARIFYING (authoritative — this outranks the " + "guidance below).\n" + f"The requirement is still being clarified and {owner} owns that. " + f"Route to {owner} unless the latest message explicitly @mentions " + "another agent by name. Never hand the floor to an implementation " + "agent on topic match alone — the specification is not settled yet, " + "so work started now would be built on guesses.\n" + ) + + _ROUTER_PROMPT = """\ You are a conversation router for a multi-agent workspace. Decide which \ agent should respond next to the LATEST message. Use judgment — read the \ @@ -646,7 +968,7 @@ def _master_targets(event, channel, mentions: List[str]) -> List[str]: Channel participants: {participants} Master agent: {master} -{plan} +{phase}{plan} Recent conversation (oldest → newest): {history} @@ -850,6 +1172,7 @@ async def _route_with_llm( prompt = _ROUTER_PROMPT.format( participants=participants_str, master=master, + phase=_phase_router_block(channel, db, workspace), plan=plan, history=history, sender=sender, @@ -1038,6 +1361,11 @@ async def _handle_message_posted(event: Event, ctx: PipelineContext) -> Optional - "next:agent-name" → route to that agent - "stop" → no targeting, conversation rests until human speaks - Fallback (single-agent threads or router disabled): no routing needed. + + Whatever the mode decides then passes through the phase gate: while the + channel is 'clarifying', only the phase owner / master keep the floor and + an explicitly @mentioned agent is downgraded to PLAN mode (see + `_apply_phase_gate`). """ from app.models import Channel, WorkspaceMember @@ -1145,6 +1473,13 @@ async def _handle_message_posted(event: Event, ctx: PipelineContext) -> Optional else: targets = _fallback_targets(event, channel, mentions, online_names) + # Requirement-clarification gate. Applied AFTER the mode picked its + # targets so it constrains every mode uniformly — the gate is the last + # word on who may be woken while the spec is unsettled. + targets, target_modes = _apply_phase_gate( + event, channel, targets, mentions, db, workspace, + ) + # ALWAYS set target_agents, even when nobody should respond. # # Use a non-empty sentinel list ["__no_response__"] instead of [] @@ -1156,6 +1491,21 @@ async def _handle_message_posted(event: Event, ctx: PipelineContext) -> Optional # and new clients to treat it as "nobody" (the sentinel is ignored). event.metadata["target_agents"] = targets if targets else ["__no_response__"] + # Phase context for the connector: `target_modes` downgrades a gated + # agent to PLAN for this message, and `phase`/`phase_owner` let every + # woken agent state the phase in its own prompt. + if _channel_phase(channel) == PHASE_CLARIFYING: + # Stamped even when no gatekeeper survives: in that state every + # target runs in plan mode, and the agent needs the phase in its + # prompt to understand why it must not build. `phase_owner` is + # simply absent then — the directive falls back to generic wording. + event.metadata["phase"] = PHASE_CLARIFYING + gatekeepers = _phase_gatekeepers(channel, db, workspace) + if gatekeepers: + event.metadata["phase_owner"] = gatekeepers[0] + if target_modes: + event.metadata["target_modes"] = target_modes + # Auto-add targeted agents as channel participants so they can poll # for messages on this channel. Three guards: # 1. Never add the `__no_response__` sentinel — it's a routing diff --git a/workspace/backend/app/routers/network.py b/workspace/backend/app/routers/network.py index 644850f14..f78e41d37 100644 --- a/workspace/backend/app/routers/network.py +++ b/workspace/backend/app/routers/network.py @@ -429,6 +429,8 @@ def discover( "master": c.master_agent, "orchestration_mode": c.orchestration_mode or "dynamic", "orchestration_instruction": c.orchestration_instruction, + "phase": c.phase or "open", + "phase_owner": c.phase_owner, "participants": [p.agent_name for p in (c.participants or [])], "created_at": created_at_ts, "last_event_at": c.last_event_at, diff --git a/workspace/backend/app/routers/workspaces.py b/workspace/backend/app/routers/workspaces.py index 1d5c9bda8..aa4c80fd2 100644 --- a/workspace/backend/app/routers/workspaces.py +++ b/workspace/backend/app/routers/workspaces.py @@ -44,8 +44,10 @@ resolve_current_user, verify_workspace_access, ) +from app.mods.workspace_mod import CHANNEL_PHASES, PHASE_CLARIFYING from app.response import ResponseCode, json_response, success_response -from app.routers.network import _workspace_filter +from app.routers.network import _emit_event_blocking, _workspace_filter +from openagents.core.onm_events import Event logger = logging.getLogger(__name__) @@ -90,6 +92,8 @@ class ChannelUpdateRequest(BaseModel): master_agent: Optional[str] = None # Reassign channel master orchestration_mode: Optional[str] = None # "dynamic" | "master" | "workflow" orchestration_instruction: Optional[str] = None # free-text plan for "workflow" mode + phase: Optional[str] = None # "open" | "clarifying" | "building" + phase_owner: Optional[str] = None # agent owning the clarifying phase ("" clears) auto_title: bool = False # When True, title update is from auto-titling (don't mark as manually set) class WorkspaceUpdateRequest(BaseModel): @@ -181,6 +185,8 @@ def _format_channel(ch: Channel) -> dict: "masterAgent": ch.master_agent, "orchestrationMode": ch.orchestration_mode or "dynamic", "orchestrationInstruction": ch.orchestration_instruction, + "phase": ch.phase or "open", + "phaseOwner": ch.phase_owner, "resumeFrom": ch.resume_from, "status": ch.status, "starred": bool(ch.starred), @@ -530,14 +536,34 @@ def remove_member( WorkspaceMember.agent_name == agent_name, ) ).scalar_one_or_none() - - if not member: + # A `removed` row is a tombstone, not a member: removal is now a soft + # delete, so without this the second DELETE would find the row left by + # the first, emit another removal event and answer 200 — losing the 404 + # this endpoint returned for an absent member back when it hard-deleted. + if not member or (member.status or "").lower() == "removed": return json_response(ResponseCode.NOT_FOUND, "Member not found") - db.delete(member) - db.commit() + # Go through the same `network.agent.remove` event as POST /v1/remove + # instead of deleting the row here. A direct delete skipped everything + # that removal has to do: leaving the `status='removed'` tombstone that + # stops a still-running daemon from re-joining (issue #347), reassigning + # the workspace and per-channel master, and handing on any clarification + # gate the agent was holding. Deleting an owner through this endpoint + # therefore left channels pointing at an agent that no longer exists. + event = Event( + type="network.agent.remove", + source="human:user", + target="core", + payload={"agent_name": agent_name}, + ) + result = _emit_event_blocking(event, workspace, db, token=workspace.password_hash) + if result is None: + return json_response(ResponseCode.NOT_FOUND, "Member not found") - return success_response({"agent_name": agent_name, "removed": True}) + resp = {"agent_name": agent_name, "removed": True} + if result.metadata.get("new_master"): + resp["new_master"] = result.metadata["new_master"] + return success_response(resp) # --------------------------------------------------------------------------- @@ -1321,6 +1347,71 @@ def update_channel( if body.orchestration_instruction is not None: # Empty string clears the plan; otherwise store the trimmed text. channel.orchestration_instruction = body.orchestration_instruction.strip() or None + # ── Clarification phase ────────────────────────────────────────── + # Phase and owner are validated together, before either is written: a + # channel must never persist `phase='clarifying'` with nobody able to + # hold the floor. That state renders as "Clarifying" in the UI while the + # gate is inert and every agent keeps answering as before — worse than + # having no gate at all, because it looks like one. + if body.phase is not None or body.phase_owner is not None: + phase = channel.phase or "open" + if body.phase is not None: + phase = body.phase.strip().lower() + if phase not in CHANNEL_PHASES: + return json_response(ResponseCode.BAD_REQUEST, "Invalid phase") + + owner = channel.phase_owner + explicit_owner = body.phase_owner is not None + if explicit_owner: + owner = body.phase_owner.strip() or None + if phase == PHASE_CLARIFYING and not owner: + # Fall back to the master, which is what a one-click "clarify + # first" means on a thread that has a leader. + owner = channel.master_agent + + def _owner_is_live(name: str) -> bool: + member = db.execute( + select(WorkspaceMember).where( + WorkspaceMember.workspace_id == workspace.id, + WorkspaceMember.agent_name == name, + ) + ).scalar_one_or_none() + return bool(member) and (member.status or "").lower() != "removed" + + if owner and not _owner_is_live(owner): + # A bad name the caller just supplied is an error. A stale one + # inherited from the row is not the caller's doing, and refusing + # the request would trap the thread: with a deleted owner, both + # "turn the gate off" and "requirement confirmed" would 400 and + # the only way out would be to appoint an owner first. Leaving + # the gate is always allowed; the dead name is cleared on the way. + if explicit_owner or phase == PHASE_CLARIFYING: + return json_response( + ResponseCode.BAD_REQUEST, + f"Unknown phase_owner: {owner}", + ) + owner = None + + if owner: + # The owner has to be able to receive messages in this channel. + # Adding them is the intent of naming them, so join them rather + # than rejecting the request. + is_participant = any( + p.agent_name == owner for p in (channel.participants or []) + ) + if not is_participant: + db.add(ChannelMember(channel_id=channel.id, agent_name=owner)) + db.flush() + + if phase == PHASE_CLARIFYING and not owner: + return json_response( + ResponseCode.BAD_REQUEST, + "phase_owner is required to start clarifying: this thread has " + "no master, so name the agent that owns the requirement", + ) + + channel.phase = phase + channel.phase_owner = owner db.commit() db.refresh(channel) diff --git a/workspace/backend/tests/test_phase_gate.py b/workspace/backend/tests/test_phase_gate.py new file mode 100644 index 000000000..2b7601e38 --- /dev/null +++ b/workspace/backend/tests/test_phase_gate.py @@ -0,0 +1,780 @@ +# -*- coding: utf-8 -*- +""" +Tests for the requirement-clarification phase gate. + +The gate exists to stop a builder agent from being handed the floor — and +starting to implement — while the requirement is still being clarified. It +runs after whatever the thread's orchestration mode decided, so these tests +drive `_handle_message_posted` end-to-end for the deterministic modes and +`_apply_phase_gate` directly for the unit-level rules. +""" + +import asyncio +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import delete, select, update + +from app.models import Channel, ChannelMember, Workspace, WorkspaceMember +from app.mods.workspace_mod import ( + PHASE_BUILDING, + PHASE_CLARIFYING, + PHASE_OPEN, + _apply_phase_gate, + _handle_agent_remove, + _handle_channel_create, + _handle_channel_leave, + _handle_message_posted, + _phase_gatekeepers, + _phase_router_block, +) +from openagents.core.onm_events import Event +from openagents.core.onm_mods import PipelineContext + + +def _make_event(source: str, content: str, target: str = "channel/session-test") -> Event: + return Event( + type="workspace.message.posted", + source=source, + target=target, + payload={"content": content, "message_type": "chat"}, + metadata={}, + ) + + +def _now(): + """Fresh heartbeat. Routing treats a member with a stale (or missing) + heartbeat as offline — see `_member_is_online` — so a fixture agent that + is meant to be reachable has to look reachable.""" + return datetime.now(timezone.utc) + + +def _run(coro): + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + + +@pytest.fixture +def gated_workspace(db): + """PM (master + phase owner) and RD in one clarifying channel.""" + ws = Workspace(name="Gate WS", slug="gate-ws", password_hash="test-token") + db.add(ws) + db.flush() + + db.add(WorkspaceMember(workspace_id=ws.id, agent_name="pm", role="master", status="online", last_heartbeat=_now())) + db.add(WorkspaceMember(workspace_id=ws.id, agent_name="rd", role="member", status="online", last_heartbeat=_now())) + db.add(WorkspaceMember(workspace_id=ws.id, agent_name="qa", role="member", status="online", last_heartbeat=_now())) + db.flush() + + ch = Channel( + workspace_id=ws.id, + name="session-test", + master_agent="pm", + phase=PHASE_CLARIFYING, + phase_owner="pm", + orchestration_mode="master", + status="active", + ) + db.add(ch) + db.flush() + for name in ("pm", "rd", "qa"): + db.add(ChannelMember(channel_id=ch.id, agent_name=name)) + db.flush() + db.refresh(ch) + return {"workspace": ws, "channel": ch} + + +class TestGatekeepers: + def test_owner_first_then_master(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = "qa" + assert _phase_gatekeepers(ch, db, ws) == ["qa", "pm"] + + def test_owner_falls_back_to_master(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = None + assert _phase_gatekeepers(ch, db, ws) == ["pm"] + + def test_no_owner_no_master_is_inert(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = None + ch.master_agent = None + assert _phase_gatekeepers(ch, db, ws) == [] + event = _make_event("human:user", "build me a thing") + # Nothing to hand the floor to: the target is kept so the human gets + # an answer, but it cannot start building. + assert _apply_phase_gate(event, ch, ["rd"], [], db, ws) == (["rd"], {"rd": "plan"}) + + +class TestGatekeeperValidation: + """A gatekeeper that cannot answer must never be routed to — that strands + the conversation with no reply at all.""" + + def test_ghost_owner_is_ignored(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = "typo-agent" + # Master survives, so the gate keeps working through it. + assert _phase_gatekeepers(ch, db, ws) == ["pm"] + + def test_unenforceable_gate_answers_but_cannot_build(self, db, gated_workspace): + """An unenforceable gate must neither eat the message nor let the + builder loose: the target is kept, but downgraded to plan.""" + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = "typo-agent" + ch.master_agent = None + event = _make_event("human:user", "write the sync code") + targets, modes = _apply_phase_gate(event, ch, ["rd"], [], db, ws) + assert targets == ["rd"] + assert modes == {"rd": "plan"} + + def test_removed_owner_is_ignored(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = "qa" + db.execute( + update(WorkspaceMember) + .where( + WorkspaceMember.workspace_id == ws.id, + WorkspaceMember.agent_name == "qa", + ) + .values(status="removed") + ) + db.flush() + assert _phase_gatekeepers(ch, db, ws) == ["pm"] + + def test_owner_that_left_the_channel_is_ignored(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = "qa" + db.execute( + delete(ChannelMember).where( + ChannelMember.channel_id == ch.id, + ChannelMember.agent_name == "qa", + ) + ) + db.flush() + db.refresh(ch) + assert _phase_gatekeepers(ch, db, ws) == ["pm"] + + def test_owner_with_a_stale_heartbeat_is_ignored(self, db, gated_workspace): + """A crashed daemon leaves status='online' in the database forever. + Handing the floor to it would make the thread look gated and healthy + while every message went unanswered.""" + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = "qa" + db.execute( + update(WorkspaceMember) + .where( + WorkspaceMember.workspace_id == ws.id, + WorkspaceMember.agent_name == "qa", + ) + .values(last_heartbeat=_now() - timedelta(hours=1)) + ) + db.flush() + assert _phase_gatekeepers(ch, db, ws) == ["pm"] + + def test_offline_owner_falls_back_to_plan_only_not_silence(self, db, gated_workspace): + """With no live gatekeeper the thread must keep answering — in plan + mode — instead of routing into a dead agent's mailbox. Ownership is + left in the database and resumes when the daemon reconnects.""" + ws = gated_workspace["workspace"] + ch = gated_workspace["channel"] + ch.master_agent = None + ch.orchestration_mode = "dynamic" + db.execute( + update(WorkspaceMember) + .where( + WorkspaceMember.workspace_id == ws.id, + WorkspaceMember.agent_name == "pm", + ) + .values(last_heartbeat=_now() - timedelta(hours=1)) + ) + db.flush() + + out = _run(_handle_message_posted( + _make_event("human:user", "can you build the sync feature?"), + PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ), + )) + assert out.metadata["target_agents"] not in (["pm"], ["__no_response__"]) + for name in out.metadata["target_agents"]: + assert out.metadata["target_modes"][name] == "plan" + db.refresh(ch) + assert ch.phase_owner == "pm", "ownership survives a disconnect" + + def test_no_valid_gatekeeper_never_targets_a_ghost(self, db, gated_workspace): + """The end-to-end shape of the bug: routing must not emit a target + that no connector will ever pick up.""" + ws = gated_workspace["workspace"] + ch = gated_workspace["channel"] + ch.phase_owner = "ghost" + ch.master_agent = None + ch.orchestration_mode = "dynamic" + db.flush() + event = _make_event("human:user", "I need an order sync feature") + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + out = _run(_handle_message_posted(event, ctx)) + assert "ghost" not in out.metadata["target_agents"] + assert out.metadata["target_agents"] != ["__no_response__"] + # Whoever picks it up must still be barred from implementing. + for name in out.metadata["target_agents"]: + assert out.metadata["target_modes"][name] == "plan" + assert out.metadata["phase"] == PHASE_CLARIFYING + + +class TestOwnerRepair: + """Membership changes must not leave a gate pointing at somebody gone.""" + + def test_removing_the_owner_hands_the_gate_to_the_master(self, db, gated_workspace): + ws = gated_workspace["workspace"] + ch = gated_workspace["channel"] + ch.phase_owner = "qa" + db.flush() + event = Event( + type="network.agent.remove", + source="human:user", + target="core", + payload={"agent_name": "qa"}, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_agent_remove(event, ctx)) + db.refresh(ch) + assert ch.phase_owner == "pm" + assert ch.phase == PHASE_CLARIFYING + + def test_removing_the_only_gatekeeper_leaves_the_gate_unowned(self, db, gated_workspace): + ws = gated_workspace["workspace"] + ch = gated_workspace["channel"] + ch.master_agent = None + db.flush() + event = Event( + type="network.agent.remove", + source="human:user", + target="core", + payload={"agent_name": "pm"}, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_agent_remove(event, ctx)) + db.refresh(ch) + # The gate stays on — losing an agent is not the user deciding the + # requirement is settled — but nobody owns it, so routing holds + # everyone in plan mode until a human names a replacement. + assert ch.phase == PHASE_CLARIFYING + assert ch.phase_owner is None + + def test_owner_who_is_also_master_leaves_nothing_stale_behind(self, db, gated_workspace): + """owner == master is the common shape. Clearing the owner is not + enough: a stale master_agent keeps every later message routed at the + agent that just walked out.""" + ws = gated_workspace["workspace"] + ch = gated_workspace["channel"] + event = Event( + type="network.channel.leave", + source="human:user", + target="core", + payload={"channel": "session-test", "agent_name": "pm"}, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_leave(event, ctx)) + db.refresh(ch) + assert ch.master_agent is None + assert ch.phase == PHASE_CLARIFYING # only a human removes a gate + assert ch.phase_owner is None + + # The next human message must reach somebody who is still here, and + # that somebody must not be free to start building. + msg = _make_event("human:user", "so where are we?") + out = _run(_handle_message_posted( + msg, + PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ), + )) + assert "pm" not in out.metadata["target_agents"] + assert out.metadata["target_agents"] != ["__no_response__"] + for name in out.metadata["target_agents"]: + assert out.metadata["target_modes"][name] == "plan" + + def test_removing_the_master_promotes_a_survivor(self, db, gated_workspace): + """Guards the query behind the repair: picking the next master over + several survivors used to raise MultipleResultsFound and abort the + whole removal.""" + ws = gated_workspace["workspace"] + event = Event( + type="network.agent.remove", + source="human:user", + target="core", + payload={"agent_name": "pm"}, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + out = _run(_handle_agent_remove(event, ctx)) + assert out.metadata["new_master"] in ("rd", "qa") + + def test_owner_leaving_the_channel_hands_the_gate_on(self, db, gated_workspace): + ws = gated_workspace["workspace"] + ch = gated_workspace["channel"] + ch.phase_owner = "qa" + db.flush() + event = Event( + type="network.channel.leave", + source="human:user", + target="core", + payload={"channel": "session-test", "agent_name": "qa"}, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_leave(event, ctx)) + db.refresh(ch) + assert ch.phase_owner == "pm" + + +class TestChannelCreateGate: + def test_clarifying_without_a_valid_owner_is_gated_but_unowned(self, db): + """Threads from the picker have no master. The gate is kept but left + unowned — routing then holds everyone in plan mode — rather than + silently creating the thread open, which would look protected while + the first message went to a builder.""" + ws = Workspace(name="Create WS", slug="create-ws", password_hash="t") + db.add(ws) + db.flush() + db.add(WorkspaceMember(workspace_id=ws.id, agent_name="rd", role="member", status="online", last_heartbeat=_now())) + db.flush() + event = Event( + type="network.channel.create", + source="human:user", + target="core", + payload={"name": "c-open", "participants": ["rd"], "phase": "clarifying"}, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_create(event, ctx)) + ch = db.execute( + select(Channel).where(Channel.workspace_id == ws.id, Channel.name == "c-open") + ).scalar_one() + assert ch.phase == PHASE_CLARIFYING, "an explicit gate request must never be dropped" + assert ch.phase_owner is None + + def test_ghost_owner_in_the_participants_payload_is_refused(self, db): + """The participants list is caller-supplied: appearing in it proves + nothing about the agent existing.""" + ws = Workspace(name="Create WS3", slug="create-ws3", password_hash="t") + db.add(ws) + db.flush() + db.add(WorkspaceMember(workspace_id=ws.id, agent_name="rd", role="member", status="online", last_heartbeat=_now())) + db.flush() + event = Event( + type="network.channel.create", + source="human:user", + target="core", + payload={ + "name": "c-ghost", + "participants": ["ghost", "rd"], + "phase": "clarifying", + "phase_owner": "ghost", + }, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_create(event, ctx)) + ch = db.execute( + select(Channel).where(Channel.workspace_id == ws.id, Channel.name == "c-ghost") + ).scalar_one() + assert ch.phase == PHASE_CLARIFYING, "an explicit gate request must never be dropped" + assert ch.phase_owner is None + + def test_owner_outside_the_participants_is_refused(self, db): + """The owner has to be in the thread it owns — the new-thread dialog + only offers selected agents, and the backend enforces the same rule.""" + ws = Workspace(name="Create WS5", slug="create-ws5", password_hash="t") + db.add(ws) + db.flush() + for name in ("pm", "rd"): + db.add(WorkspaceMember(workspace_id=ws.id, agent_name=name, role="member", status="online", last_heartbeat=_now())) + db.flush() + event = Event( + type="network.channel.create", + source="human:user", + target="core", + payload={ + "name": "c-outsider", + "participants": ["rd"], # pm was not selected + "phase": "clarifying", + "phase_owner": "pm", + }, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_create(event, ctx)) + ch = db.execute( + select(Channel).where(Channel.workspace_id == ws.id, Channel.name == "c-outsider") + ).scalar_one() + assert ch.phase == PHASE_CLARIFYING, "an explicit gate request must never be dropped" + assert ch.phase_owner is None + + def test_creating_without_a_phase_is_open(self, db): + """Unchecking "clarify first" sends no phase at all.""" + ws = Workspace(name="Create WS6", slug="create-ws6", password_hash="t") + db.add(ws) + db.flush() + for name in ("pm", "rd"): + db.add(WorkspaceMember(workspace_id=ws.id, agent_name=name, role="member", status="online", last_heartbeat=_now())) + db.flush() + event = Event( + type="network.channel.create", + source="human:user", + target="core", + payload={"name": "c-plain", "participants": ["pm", "rd"]}, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_create(event, ctx)) + ch = db.execute( + select(Channel).where(Channel.workspace_id == ws.id, Channel.name == "c-plain") + ).scalar_one() + assert ch.phase == PHASE_OPEN + assert ch.phase_owner is None + + def test_create_reports_what_was_persisted(self, db): + """Clients must render the gate from the response, not from what they + asked for — the two differ exactly when the owner went stale.""" + ws = Workspace(name="Create WS8", slug="create-ws8", password_hash="t") + db.add(ws) + db.flush() + db.add(WorkspaceMember(workspace_id=ws.id, agent_name="rd", role="member", status="online", last_heartbeat=_now())) + db.flush() + event = Event( + type="network.channel.create", + source="human:user", + target="core", + payload={ + "name": "c-report", + "participants": ["rd"], + "phase": "clarifying", + "phase_owner": "pm-that-just-left", + }, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + out = _run(_handle_channel_create(event, ctx)) + assert out.metadata["phase"] == PHASE_CLARIFYING + assert out.metadata["phase_owner"] is None + + def test_unowned_gate_from_creation_still_blocks_building(self, db): + """The safety claim behind keeping an unowned gate: the first message + is answered, but nobody may implement.""" + ws = Workspace(name="Create WS9", slug="create-ws9", password_hash="t") + db.add(ws) + db.flush() + for name in ("pm", "rd"): + db.add(WorkspaceMember(workspace_id=ws.id, agent_name=name, role="member", status="online", last_heartbeat=_now())) + db.flush() + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_create(Event( + type="network.channel.create", + source="human:user", + target="core", + payload={ + "name": "c-unowned", + "participants": ["rd", "pm"], + "phase": "clarifying", + "phase_owner": "gone", + }, + metadata={}, + ), ctx)) + + out = _run(_handle_message_posted( + _make_event("human:user", "ship the sync feature", target="channel/c-unowned"), + ctx, + )) + assert out.metadata["target_agents"] != ["__no_response__"] + for name in out.metadata["target_agents"]: + assert out.metadata["target_modes"][name] == "plan" + + def test_gate_is_live_for_the_very_first_message(self, db): + """The whole point of sending the phase with the create event: there + must be no window in which the thread exists ungated and the first + request can be routed to a builder.""" + ws = Workspace(name="Create WS7", slug="create-ws7", password_hash="t") + db.add(ws) + db.flush() + for name in ("pm", "rd"): + db.add(WorkspaceMember(workspace_id=ws.id, agent_name=name, role="member", status="online", last_heartbeat=_now())) + db.flush() + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_create(Event( + type="network.channel.create", + source="human:user", + target="core", + payload={ + # rd first, so the ungated fallback would pick rd — the + # assertion below only holds if the gate is already live. + "name": "c-first", + "participants": ["rd", "pm"], + "phase": "clarifying", + "phase_owner": "pm", + }, + metadata={}, + ), ctx)) + + # No PATCH in between — straight to the opening request. + first = _make_event( + "human:user", "build me an order sync feature", target="channel/c-first", + ) + out = _run(_handle_message_posted(first, ctx)) + assert out.metadata["target_agents"] == ["pm"] + assert out.metadata["phase"] == PHASE_CLARIFYING + assert out.metadata["phase_owner"] == "pm" + + def test_removed_owner_at_creation_is_refused(self, db): + ws = Workspace(name="Create WS4", slug="create-ws4", password_hash="t") + db.add(ws) + db.flush() + db.add(WorkspaceMember(workspace_id=ws.id, agent_name="pm", role="member", status="removed", last_heartbeat=_now())) + db.add(WorkspaceMember(workspace_id=ws.id, agent_name="rd", role="member", status="online", last_heartbeat=_now())) + db.flush() + event = Event( + type="network.channel.create", + source="human:user", + target="core", + payload={ + "name": "c-removed", + "participants": ["pm", "rd"], + "phase": "clarifying", + "phase_owner": "pm", + }, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_create(event, ctx)) + ch = db.execute( + select(Channel).where(Channel.workspace_id == ws.id, Channel.name == "c-removed") + ).scalar_one() + assert ch.phase == PHASE_CLARIFYING, "an explicit gate request must never be dropped" + assert ch.phase_owner is None + + def test_clarifying_with_a_participant_owner_is_honoured(self, db): + ws = Workspace(name="Create WS2", slug="create-ws2", password_hash="t") + db.add(ws) + db.flush() + for name in ("pm", "rd"): + db.add(WorkspaceMember(workspace_id=ws.id, agent_name=name, role="member", status="online", last_heartbeat=_now())) + db.flush() + event = Event( + type="network.channel.create", + source="human:user", + target="core", + payload={ + "name": "c-gated", + "participants": ["pm", "rd"], + "phase": "clarifying", + "phase_owner": "pm", + }, + metadata={}, + ) + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + _run(_handle_channel_create(event, ctx)) + ch = db.execute( + select(Channel).where(Channel.workspace_id == ws.id, Channel.name == "c-gated") + ).scalar_one() + assert ch.phase == PHASE_CLARIFYING + assert ch.phase_owner == "pm" + + +class TestApplyPhaseGate: + def test_open_phase_is_a_noop(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase = PHASE_OPEN + event = _make_event("human:user", "build me a thing") + assert _apply_phase_gate(event, ch, ["rd"], [], db, ws) == (["rd"], {}) + + def test_building_phase_is_a_noop(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase = PHASE_BUILDING + event = _make_event("human:user", "build me a thing") + assert _apply_phase_gate(event, ch, ["rd"], [], db, ws) == (["rd"], {}) + + def test_unmentioned_builder_is_redirected_to_the_owner(self, db, gated_workspace): + """The reported bug: the router picks RD on topic match while the + requirement is still being clarified.""" + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + event = _make_event("human:user", "I want a feature that syncs orders") + targets, modes = _apply_phase_gate(event, ch, ["rd"], [], db, ws) + assert targets == ["pm"] + assert modes == {} + + def test_mentioned_builder_is_kept_but_downgraded_to_plan(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + event = _make_event("human:user", "@rd is this feasible at all?") + targets, modes = _apply_phase_gate(event, ch, ["rd"], ["rd"], db, ws) + assert targets == ["rd"] + assert modes == {"rd": "plan"} + + def test_owner_delegating_by_mention_gets_plan_mode(self, db, gated_workspace): + """PM consulting RD mid-clarification must not start implementation.""" + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + event = _make_event("openagents:pm", "@rd how long would the sync take?") + targets, modes = _apply_phase_gate(event, ch, ["rd"], ["rd"], db, ws) + assert targets == ["rd"] + assert modes == {"rd": "plan"} + + def test_gatekeeper_target_passes_through(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + event = _make_event("openagents:rd", "here's my read on feasibility") + assert _apply_phase_gate(event, ch, ["pm"], [], db, ws) == (["pm"], {}) + + def test_owner_own_message_does_not_self_loop(self, db, gated_workspace): + """Everything dropped and the owner is the sender → end the turn + instead of routing the owner back to itself.""" + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + event = _make_event("openagents:pm", "so, a couple of questions for you") + targets, modes = _apply_phase_gate(event, ch, ["rd"], [], db, ws) + assert targets == [] + assert modes == {} + + def test_second_gatekeeper_takes_over_when_owner_is_the_sender(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = "qa" + event = _make_event("openagents:qa", "still gathering requirements") + targets, _ = _apply_phase_gate(event, ch, ["rd"], [], db, ws) + assert targets == ["pm"] + + def test_mention_of_an_untargeted_agent_does_not_add_it(self, db, gated_workspace): + """`mentions` only whitelists agents the mode already targeted — the + gate narrows routing, it never widens it.""" + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + event = _make_event("human:user", "@qa what do you think about rd's plan?") + targets, modes = _apply_phase_gate(event, ch, ["rd"], ["qa"], db, ws) + assert targets == ["pm"] + assert modes == {} + + +class TestRouterPromptBlock: + def test_block_names_the_owner_when_clarifying(self, db, gated_workspace): + block = _phase_router_block(gated_workspace["channel"], db, gated_workspace["workspace"]) + assert "CLARIFYING" in block + assert "pm" in block + + def test_block_is_empty_when_not_clarifying(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase = PHASE_OPEN + assert _phase_router_block(ch, db, ws) == "" + + def test_block_is_empty_without_gatekeepers(self, db, gated_workspace): + ch = gated_workspace["channel"] + ws = gated_workspace["workspace"] + ch.phase_owner = None + ch.master_agent = None + assert _phase_router_block(ch, db, ws) == "" + + +class TestEndToEnd: + """Through `_handle_message_posted` in master mode (no LLM involved).""" + + def test_master_mode_delegation_to_builder_is_plan_gated(self, db, gated_workspace): + ws = gated_workspace["workspace"] + event = _make_event("openagents:pm", "@rd please start on the sync module") + ctx = PipelineContext( + network_id=str(ws.id), agent_address="openagents:pm", db=db, workspace=ws, + ) + out = _run(_handle_message_posted(event, ctx)) + assert out.metadata["target_agents"] == ["rd"] + assert out.metadata["target_modes"] == {"rd": "plan"} + assert out.metadata["phase"] == PHASE_CLARIFYING + assert out.metadata["phase_owner"] == "pm" + + def test_human_message_reaches_the_owner(self, db, gated_workspace): + ws = gated_workspace["workspace"] + event = _make_event("human:user", "I need an order sync feature") + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + out = _run(_handle_message_posted(event, ctx)) + assert out.metadata["target_agents"] == ["pm"] + assert "target_modes" not in out.metadata + + def test_building_phase_restores_normal_delegation(self, db, gated_workspace): + ws = gated_workspace["workspace"] + ch = gated_workspace["channel"] + ch.phase = PHASE_BUILDING + db.flush() + event = _make_event("openagents:pm", "@rd please start on the sync module") + ctx = PipelineContext( + network_id=str(ws.id), agent_address="openagents:pm", db=db, workspace=ws, + ) + out = _run(_handle_message_posted(event, ctx)) + assert out.metadata["target_agents"] == ["rd"] + assert "target_modes" not in out.metadata + assert "phase" not in out.metadata + + def test_gated_thread_never_leaves_a_human_unanswered(self, db, gated_workspace): + """A human message that would have gone to a builder still gets a + reply — from the owner.""" + ws = gated_workspace["workspace"] + ch = gated_workspace["channel"] + ch.orchestration_mode = "dynamic" + # No master and no router key → the fallback picks the first + # participant (pm), who is NOT the gatekeeper here, so the gate has to + # redirect rather than drop the message. + ch.master_agent = None + ch.phase_owner = "qa" + db.flush() + event = _make_event("human:user", "write the code for order sync") + ctx = PipelineContext( + network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws, + ) + out = _run(_handle_message_posted(event, ctx)) + assert out.metadata["target_agents"] == ["qa"] diff --git a/workspace/backend/tests/test_workspaces.py b/workspace/backend/tests/test_workspaces.py index 7a7f87ecc..5815dbce8 100644 --- a/workspace/backend/tests/test_workspaces.py +++ b/workspace/backend/tests/test_workspaces.py @@ -4,6 +4,7 @@ """ import pytest +from sqlalchemy import select from unittest.mock import patch, MagicMock @@ -233,6 +234,175 @@ def test_empty_instruction_clears_plan(self, client, workspace): assert resp.json()["data"]["orchestrationInstruction"] is None +class TestChannelPhase: + """PATCH /v1/workspaces/{id}/channels/{name} — clarification phase gate.""" + + def _patch(self, client, workspace, body): + return client.patch( + f"/v1/workspaces/{workspace['id']}/channels/{workspace['channel']['name']}", + json=body, + headers={"X-Workspace-Token": workspace["token"]}, + ) + + def _join(self, client, workspace, name): + resp = client.post("/v1/join", json={ + "agent_name": name, + "token": workspace["token"], + "network": workspace["id"], + }) + assert resp.status_code == 200 + + def test_default_phase_is_open(self, client, workspace): + assert workspace["channel"].get("phase") == "open" + + def test_clarifying_defaults_owner_to_master(self, client, workspace): + """A one-click "start clarifying" must produce a working gate, not an + inert one with nobody holding the floor.""" + resp = self._patch(client, workspace, {"phase": "clarifying"}) + assert resp.status_code == 200 + data = resp.json()["data"] + assert data["phase"] == "clarifying" + assert data["phaseOwner"] == data["masterAgent"] == "agent-alpha" + + def test_explicit_owner_wins(self, client, workspace): + self._join(client, workspace, "agent-pm") + resp = self._patch(client, workspace, { + "phase": "clarifying", + "phase_owner": "agent-pm", + }) + assert resp.status_code == 200 + assert resp.json()["data"]["phaseOwner"] == "agent-pm" + + def test_owner_is_joined_to_the_channel(self, client, workspace): + """Naming an agent as owner means it must be able to receive the + thread's messages — otherwise the gate routes into a void.""" + self._join(client, workspace, "agent-pm") + resp = self._patch(client, workspace, { + "phase": "clarifying", + "phase_owner": "agent-pm", + }) + assert "agent-pm" in resp.json()["data"]["participants"] + + def test_unknown_owner_rejected(self, client, workspace): + resp = self._patch(client, workspace, { + "phase": "clarifying", + "phase_owner": "typo-agent", + }) + assert resp.status_code == 400 + + def test_removed_owner_rejected(self, client, workspace): + self._join(client, workspace, "agent-pm") + client.delete( + f"/v1/workspaces/{workspace['id']}/members/agent-pm", + headers={"X-Workspace-Token": workspace["token"]}, + ) + resp = self._patch(client, workspace, { + "phase": "clarifying", + "phase_owner": "agent-pm", + }) + assert resp.status_code == 400 + + def test_clarifying_without_any_owner_rejected(self, client, workspace, db): + """Threads created from the agent picker have no master. Entering the + gate there used to persist phase=clarifying with owner=None, which + renders as "Clarifying" while routing stays wide open.""" + from app.models import Channel + channel = db.execute( + select(Channel).where(Channel.name == workspace["channel"]["name"]) + ).scalar_one() + channel.master_agent = None + db.commit() + + resp = self._patch(client, workspace, {"phase": "clarifying"}) + assert resp.status_code == 400 + db.refresh(channel) + assert channel.phase == "open" + assert channel.phase_owner is None + + def test_clearing_the_owner_of_a_gated_thread_falls_back_to_master(self, client, workspace): + """Clearing the owner is allowed while the master can still hold the + floor — the gate stays enforceable, which is the only invariant.""" + self._join(client, workspace, "agent-pm") + self._patch(client, workspace, {"phase": "clarifying", "phase_owner": "agent-pm"}) + resp = self._patch(client, workspace, {"phase_owner": " "}) + assert resp.status_code == 200 + assert resp.json()["data"]["phaseOwner"] == "agent-alpha" # the master + + def test_clearing_the_only_owner_of_a_gated_thread_rejected(self, client, workspace, db): + from app.models import Channel + self._patch(client, workspace, {"phase": "clarifying"}) + channel = db.execute( + select(Channel).where(Channel.name == workspace["channel"]["name"]) + ).scalar_one() + channel.master_agent = None + db.commit() + + resp = self._patch(client, workspace, {"phase_owner": " "}) + assert resp.status_code == 400 + db.refresh(channel) + assert channel.phase_owner == "agent-alpha" + + def test_clearing_the_owner_is_fine_once_open(self, client, workspace): + self._patch(client, workspace, {"phase": "clarifying"}) + self._patch(client, workspace, {"phase": "open"}) + resp = self._patch(client, workspace, {"phase_owner": " "}) + assert resp.status_code == 200 + assert resp.json()["data"]["phaseOwner"] is None + + @pytest.mark.parametrize("target_phase", ["open", "building"]) + def test_stale_owner_does_not_trap_the_thread(self, client, workspace, db, target_phase): + """A legacy/orphaned owner must not make "Turn the gate off" and + "Requirement confirmed" both impossible — that leaves the user stuck + inside a gate they cannot leave.""" + from app.models import Channel + self._patch(client, workspace, {"phase": "clarifying"}) + channel = db.execute( + select(Channel).where(Channel.name == workspace["channel"]["name"]) + ).scalar_one() + channel.phase_owner = "ghost" + db.commit() + + resp = self._patch(client, workspace, {"phase": target_phase}) + assert resp.status_code == 200 + data = resp.json()["data"] + assert data["phase"] == target_phase + assert data["phaseOwner"] is None # the dead name is cleared on the way out + + def test_stale_owner_still_blocks_staying_in_the_gate(self, client, workspace, db): + from app.models import Channel + self._patch(client, workspace, {"phase": "clarifying"}) + channel = db.execute( + select(Channel).where(Channel.name == workspace["channel"]["name"]) + ).scalar_one() + channel.phase_owner = "ghost" + channel.master_agent = None + db.commit() + + resp = self._patch(client, workspace, {"phase": "clarifying"}) + assert resp.status_code == 400 + + def test_advance_to_building(self, client, workspace): + self._patch(client, workspace, {"phase": "clarifying"}) + resp = self._patch(client, workspace, {"phase": "building"}) + assert resp.status_code == 200 + assert resp.json()["data"]["phase"] == "building" + + def test_invalid_phase_rejected(self, client, workspace): + resp = self._patch(client, workspace, {"phase": "shipping"}) + assert resp.status_code == 400 + + def test_phase_round_trips_via_get(self, client, workspace): + self._join(client, workspace, "agent-pm") + self._patch(client, workspace, {"phase": "clarifying", "phase_owner": "agent-pm"}) + got = client.get( + f"/v1/workspaces/{workspace['id']}/channels/{workspace['channel']['name']}", + headers={"X-Workspace-Token": workspace["token"]}, + ) + data = got.json()["data"] + assert data["phase"] == "clarifying" + assert data["phaseOwner"] == "agent-pm" + + class TestGenerateMemberDescription: """POST /v1/workspaces/{id}/members/{name}/generate-description.""" @@ -435,6 +605,55 @@ def test_remove_member(self, client, workspace): names = [a["address"] for a in disc.json()["data"]["agents"]] assert "openagents:agent-to-remove" not in names + def test_remove_leaves_a_tombstone_and_repairs_the_gate(self, client, workspace, db): + """This endpoint used to hard-delete the row, skipping the removal + handler: no `removed` tombstone (so the agent's daemon could re-join), + no master reassignment, and no repair of a clarification gate the + agent was holding.""" + from app.models import Channel, WorkspaceMember + + self_join = client.post("/v1/join", json={ + "agent_name": "agent-pm", + "token": workspace["token"], + "network": workspace["id"], + }) + assert self_join.status_code == 200 + channel_name = workspace["channel"]["name"] + client.patch( + f"/v1/workspaces/{workspace['id']}/channels/{channel_name}", + json={"phase": "clarifying", "phase_owner": "agent-pm"}, + headers={"X-Workspace-Token": workspace["token"]}, + ) + + resp = client.delete( + f"/v1/workspaces/{workspace['id']}/members/agent-pm", + headers={"X-Workspace-Token": workspace["token"]}, + ) + assert resp.status_code == 200 + + member = db.execute( + select(WorkspaceMember).where(WorkspaceMember.agent_name == "agent-pm") + ).scalar_one() + assert member.status == "removed", "tombstone must survive so a stale daemon can't re-join" + + channel = db.execute( + select(Channel).where(Channel.name == channel_name) + ).scalar_one() + assert channel.phase_owner != "agent-pm" + + def test_removing_twice_returns_404(self, client, workspace): + """The tombstone left by the first removal is not a member — the + endpoint's 404-for-absent contract has to survive soft deletion.""" + client.post("/v1/join", json={ + "agent_name": "agent-twice", + "token": workspace["token"], + "network": workspace["id"], + }) + url = f"/v1/workspaces/{workspace['id']}/members/agent-twice" + headers = {"X-Workspace-Token": workspace["token"]} + assert client.delete(url, headers=headers).status_code == 200 + assert client.delete(url, headers=headers).status_code == 404 + def test_remove_nonexistent_member(self, client, workspace): """Removing nonexistent member returns 404.""" resp = client.delete( diff --git a/workspace/frontend/components/chat/chat-view.tsx b/workspace/frontend/components/chat/chat-view.tsx index 65f9b9f5b..c75e295ab 100644 --- a/workspace/frontend/components/chat/chat-view.tsx +++ b/workspace/frontend/components/chat/chat-view.tsx @@ -21,6 +21,7 @@ import { import { ListTree, MessageSquare, CalendarClock, Square, ChevronLeft, X, Plus, Globe, Share2, Crown, AlertTriangle, Sparkles } from 'lucide-react'; import { ShareDialog } from './share-dialog'; import { OrchestrationControl } from './orchestration-control'; +import { PhaseControl } from './phase-control'; import { useLayout } from '@/components/layout/layout-context'; import { DetailHeader } from '@/components/layout/app-header'; import { cn } from '@/lib/utils'; @@ -708,17 +709,31 @@ export function ChatView() { )} - {/* Orchestration mode picker — only for multi-agent threads */} + {/* Orchestration mode picker — only for multi-agent threads. + The phase control also shows on a thread that dropped below two + agents while gated: that is exactly the state a human has to + repair (pick a new owner, or release the gate), and hiding the + control would leave no way to do it. */} {!isDM && currentSession && (() => { const participants = currentSession.participants || []; const sessionAgents = agents.filter((a) => participants.includes(a.agentName)); - if (sessionAgents.length < 2) return null; + const isGated = (currentSession.phase || 'open') === 'clarifying'; + if (sessionAgents.length < 2 && !isGated) return null; return ( - setSessionOrchestration(currentSessionId!, updates)} - /> + <> + setSessionOrchestration(currentSessionId!, updates)} + /> + {sessionAgents.length >= 2 && ( + setSessionOrchestration(currentSessionId!, updates)} + /> + )} + ); })()} diff --git a/workspace/frontend/components/chat/phase-control.tsx b/workspace/frontend/components/chat/phase-control.tsx new file mode 100644 index 000000000..b4e35be81 --- /dev/null +++ b/workspace/frontend/components/chat/phase-control.tsx @@ -0,0 +1,173 @@ +'use client'; + +import * as React from 'react'; +import { ClipboardCheck, Hammer, Check, Crown, AlertTriangle } from 'lucide-react'; +import { Button } from '@/components/ui/button'; +import { + DropdownMenu, + DropdownMenuTrigger, + DropdownMenuContent, + DropdownMenuItem, + DropdownMenuLabel, + DropdownMenuSeparator, +} from '@/components/ui/dropdown-menu'; +import { cn } from '@/lib/utils'; +import { useT } from '@/lib/i18n'; +import type { WorkspaceSession, WorkspaceAgent } from '@/lib/types'; + +type Phase = 'open' | 'clarifying' | 'building'; + +interface Props { + session: WorkspaceSession; + agents: WorkspaceAgent[]; + onChange: (updates: { phase?: Phase; phaseOwner?: string | null }) => void; +} + +/** + * Header control for the requirement-clarification gate. + * + * While a thread is `clarifying`, the backend keeps routing with the phase + * owner: no other agent can be handed the floor on topic match, and one that + * is explicitly @mentioned answers in plan mode instead of starting to build. + * This is the release valve — the user says when the requirement is settled. + * + * Turning the gate ON always names an owner in the same request. Threads + * created from the agent picker deliberately have no master, so a bare + * `phase: 'clarifying'` would leave the backend with nobody to hold the + * floor: it rejects that, and this control would otherwise have shown + * "Clarifying" over routing that never changed. + * + * Off ('open') by default, so a thread only behaves this way once someone + * asks it to. + */ +export function PhaseControl({ session, agents, onChange }: Props) { + const t = useT(); + const phase = (session.phase || 'open') as Phase; + // Resolve the owner against the agents actually in this thread. A stale + // name left behind by a deleted agent is a non-empty string, so trusting + // truthiness would render a confident "Clarifying · @ghost" over routing + // that has already fallen back to the master — or to the plan-safe path + // where nobody holds the floor at all. + const known = (name: string | null) => + !!name && agents.some((a) => a.agentName === name); + const owner = known(session.phaseOwner) + ? session.phaseOwner + : known(session.master) + ? session.master + : null; + + const ownerMenu = (label: string) => ( + <> + {label} +

+ {t('phaseGate.ownerHint')} +

+ + {agents.map((a) => ( + { + e.preventDefault(); + // Phase and owner travel together: the backend refuses a gate + // with nobody able to hold it. + onChange({ phase: 'clarifying', phaseOwner: a.agentName }); + }} + className="flex items-center gap-2 py-1.5 text-xs cursor-pointer" + > + @{a.agentName} + {a.role === 'master' && } + {a.agentName === owner && } + + ))} + + ); + + if (phase === 'clarifying') { + // Legacy rows written before the owner became mandatory, or an owner that + // was removed between renders. Say so instead of implying a gate that + // isn't being enforced. + const ownerless = !owner; + return ( +
+ + + + + + {ownerMenu(t('phaseGate.ownerMenuLabel'))} + + { + e.preventDefault(); + onChange({ phase: 'open' }); + }} + className="text-xs cursor-pointer" + > + {t('phaseGate.turnOff')} + + + + + +
+ ); + } + + return ( + + + + + + {ownerMenu(t('phaseGate.clarifyFirstMenuLabel'))} + + + ); +} diff --git a/workspace/frontend/components/threads/new-thread-dialog-host.tsx b/workspace/frontend/components/threads/new-thread-dialog-host.tsx index ca6ae56bf..84f0c1ab1 100644 --- a/workspace/frontend/components/threads/new-thread-dialog-host.tsx +++ b/workspace/frontend/components/threads/new-thread-dialog-host.tsx @@ -20,8 +20,8 @@ export function NewThreadDialogHost() { onOpenChange={setNewThreadOpen} agents={agents} sessions={sessions} - onCreateThread={({ participants, resumeFrom }) => { - createSession({ participants, resumeFrom }); + onCreateThread={({ participants, resumeFrom, phase, phaseOwner }) => { + createSession({ participants, resumeFrom, phase, phaseOwner }); setViewMode('threads'); // On mobile, jump to the detail pane so the new thread is visible. if (isMobile) openMobileDetail(); diff --git a/workspace/frontend/components/threads/new-thread-dialog.tsx b/workspace/frontend/components/threads/new-thread-dialog.tsx index 0a7bfb304..4a482d56b 100644 --- a/workspace/frontend/components/threads/new-thread-dialog.tsx +++ b/workspace/frontend/components/threads/new-thread-dialog.tsx @@ -13,7 +13,7 @@ import { } from '@/components/ui/responsive-dialog'; import { Button } from '@/components/ui/button'; import { cn } from '@/lib/utils'; -import { History, Check, Minus, Users } from 'lucide-react'; +import { History, Check, Minus, Users, ClipboardCheck } from 'lucide-react'; import type { WorkspaceAgent, WorkspaceSession } from '@/lib/types'; import { AgentAvatar } from '@/components/agents/agent-avatar'; import { @@ -37,7 +37,25 @@ interface NewThreadDialogProps { onOpenChange: (open: boolean) => void; agents: WorkspaceAgent[]; sessions?: WorkspaceSession[]; - onCreateThread: (opts: { participants: string[]; resumeFrom?: string }) => void; + onCreateThread: (opts: { + participants: string[]; + resumeFrom?: string; + phase?: string; + phaseOwner?: string; + }) => void; +} + +/** + * Who should own the clarification phase for a set of selected agents. + * + * The workspace master is the only coordinator signal the system actually + * has, so it is preselected when it is among the participants. Otherwise the + * user picks: guessing would often land on the agent that builds things, and + * a gate whose owner is the builder is no gate at all. + */ +function defaultClarifyOwner(agents: WorkspaceAgent[], selected: Set): string { + const master = agents.find((a) => selected.has(a.agentName) && a.role === 'master'); + return master ? master.agentName : ''; } export function NewThreadDialog({ open, onOpenChange, agents, sessions, onCreateThread }: NewThreadDialogProps) { @@ -49,6 +67,12 @@ export function NewThreadDialog({ open, onOpenChange, agents, sessions, onCreate const [selected, setSelected] = useState>(new Set()); const [resumeFrom, setResumeFrom] = useState(NO_RESUME); + // Multi-agent threads start in clarification by default: without it the + // router hands the first underspecified request straight to whichever agent + // matches the topic, which is how an RD agent ends up writing code before + // the requirement exists. + const [clarifyFirst, setClarifyFirst] = useState(true); + const [clarifyOwner, setClarifyOwner] = useState(''); const isAllSelected = onlineAgents.length > 0 && selected.size === onlineAgents.length; const isPartiallySelected = selected.size > 0 && selected.size < onlineAgents.length; @@ -61,11 +85,35 @@ export function NewThreadDialog({ open, onOpenChange, agents, sessions, onCreate // pre-select it so the common single-agent case is a one-click "Start Thread". useEffect(() => { if (open) { - setSelected(onlineAgents.length === 1 ? new Set([onlineAgents[0].agentName]) : new Set()); + const initial = onlineAgents.length === 1 + ? new Set([onlineAgents[0].agentName]) + : new Set(); + setSelected(initial); setResumeFrom(NO_RESUME); + setClarifyFirst(true); + setClarifyOwner(defaultClarifyOwner(onlineAgents, initial)); } }, [open]); // eslint-disable-line react-hooks/exhaustive-deps + // The agents that would actually join: `selected` can name an agent that + // has since gone offline, and `handleCreate` filters participants against + // the CURRENT online list. Everything the gate decides — whether to offer + // it, who may own it, whether the owner is still valid — has to be derived + // from this same set, or the dialog offers a gate whose owner it then omits + // from the participants it submits. + const selectedOnline = agentNames.filter((n) => selected.has(n)); + const selectedOnlineKey = selectedOnline.join(','); + + // Re-derive the owner whenever that set changes — deselecting the chosen + // owner, or discovery reporting it offline, must not leave a stale name + // behind for submit. + useEffect(() => { + const live = new Set(selectedOnline); + setClarifyOwner((prev) => + prev && live.has(prev) ? prev : defaultClarifyOwner(onlineAgents, live) + ); + }, [selectedOnlineKey]); // eslint-disable-line react-hooks/exhaustive-deps + const toggleAgent = (name: string) => { setSelected((prev) => { const next = new Set(prev); @@ -78,12 +126,41 @@ export function NewThreadDialog({ open, onOpenChange, agents, sessions, onCreate }); }; + // A gate only means anything once two agents can compete for the floor. + // Offer it on the user's INTENT (what they selected), not on who happens to + // be online right now: deriving visibility from the live set alone meant an + // agent going offline made the whole option disappear mid-dialog and the + // thread was created ungated without anyone saying so. + const showClarifyOption = selected.size >= 2; + // Whether it can actually be applied is a separate question, and it is the + // live set that answers it. + const canGate = selectedOnline.length >= 2; + // A selected agent dropped off between opening the dialog and now. Creating + // regardless would silently produce something the user did not ask for — a + // smaller thread, and an ungated one. Make them resolve it. + const droppedOffline = selected.size - selectedOnline.length; + const gateBlocked = showClarifyOption && clarifyFirst && !canGate; + // Checked but no valid owner: the thread would be created gated-but-unowned + // (plan-only for everyone), so block here instead of shipping that state. + const needsOwner = + showClarifyOption && canGate && clarifyFirst && !selectedOnline.includes(clarifyOwner); + // Never create an empty thread: every selected agent may have gone offline. + const nothingToCreate = selectedOnline.length === 0; + const handleCreate = () => { // No leader is assigned at creation — the default "dynamic" mode doesn't // need one. A leader can be set later from the thread's agent menu (and is // only required by "master" orchestration mode). - const participants = agentNames.filter((n) => selected.has(n)); - onCreateThread({ participants, resumeFrom: resumeFrom === NO_RESUME ? undefined : resumeFrom }); + const participants = selectedOnline; + // The phase travels with the create event rather than a PATCH afterwards: + // a thread that is ungated for even a moment can have its first message + // routed to a builder before the gate lands. + const gated = canGate && clarifyFirst && !!clarifyOwner; + onCreateThread({ + participants, + resumeFrom: resumeFrom === NO_RESUME ? undefined : resumeFrom, + ...(gated && { phase: 'clarifying', phaseOwner: clarifyOwner }), + }); onOpenChange(false); }; @@ -196,6 +273,78 @@ export function NewThreadDialog({ open, onOpenChange, agents, sessions, onCreate )} + {/* Clarify-first gate — multi-agent threads only */} + {showClarifyOption && ( +
+ + + {clarifyFirst && gateBlocked && ( +
+

+ {droppedOffline > 0 + ? t('newThread.gateBlockedOffline', { count: droppedOffline }) + : t('newThread.gateBlockedTooFew')} +

+
+ )} + + {clarifyFirst && !gateBlocked && ( +
+ + + {needsOwner && ( +

+ {t('newThread.clarifyOwnerRequired')} +

+ )} +
+ )} +
+ )} + {/* Resume from past session — show when there are resumable sessions */} {hasClaudeAgent && resumableSessions.length > 0 && (
@@ -226,7 +375,11 @@ export function NewThreadDialog({ open, onOpenChange, agents, sessions, onCreate - diff --git a/workspace/frontend/lib/api.ts b/workspace/frontend/lib/api.ts index cc1c2ee55..096c3d07b 100644 --- a/workspace/frontend/lib/api.ts +++ b/workspace/frontend/lib/api.ts @@ -289,13 +289,16 @@ class WorkspaceApi { }); } - async updateChannel(channelName: string, updates: { title?: string; status?: string; starred?: boolean; masterAgent?: string; orchestrationMode?: string; orchestrationInstruction?: string | null }): Promise { + async updateChannel(channelName: string, updates: { title?: string; status?: string; starred?: boolean; masterAgent?: string; orchestrationMode?: string; orchestrationInstruction?: string | null; phase?: string; phaseOwner?: string | null }): Promise { // Map camelCase fields → snake_case for the backend. - const { masterAgent, orchestrationMode, orchestrationInstruction, ...rest } = updates; + const { masterAgent, orchestrationMode, orchestrationInstruction, phase, phaseOwner, ...rest } = updates; const body: Record = { ...rest }; if (masterAgent !== undefined) body.master_agent = masterAgent; if (orchestrationMode !== undefined) body.orchestration_mode = orchestrationMode; if (orchestrationInstruction !== undefined) body.orchestration_instruction = orchestrationInstruction; + if (phase !== undefined) body.phase = phase; + // null clears the owner; the backend reads an empty string as "clear". + if (phaseOwner !== undefined) body.phase_owner = phaseOwner ?? ''; return this.request(`/v1/workspaces/${this.workspaceId}/channels/${channelName}`, { method: 'PATCH', body: JSON.stringify(body), @@ -326,6 +329,8 @@ class WorkspaceApi { master?: string; participants?: string[]; resumeFrom?: string; + phase?: string; + phaseOwner?: string; } = {}): Promise { const event = await this.sendEvent({ type: 'network.channel.create', @@ -336,6 +341,11 @@ class WorkspaceApi { ...(opts.master && { master: opts.master }), ...(opts.participants && { participants: opts.participants }), ...(opts.resumeFrom && { resume_from: opts.resumeFrom }), + // Sent with the create event, never PATCHed afterwards: a thread that + // is ungated for even a moment can have its first message routed to a + // builder before the gate lands. + ...(opts.phase && { phase: opts.phase }), + ...(opts.phaseOwner && { phase_owner: opts.phaseOwner }), }, }); @@ -352,6 +362,11 @@ class WorkspaceApi { master: opts.master || null, orchestrationMode: 'dynamic', orchestrationInstruction: null, + // Read back what the backend persisted, never what we asked for: an + // owner can go stale between listing the agents and the create landing, + // and rendering the request would claim a gate the thread may not have. + phase: (event.metadata?.phase as string) || 'open', + phaseOwner: (event.metadata?.phase_owner as string) ?? null, createdAt: new Date(event.timestamp).toISOString(), lastEventAt: null, }; diff --git a/workspace/frontend/lib/i18n/messages/en-US.ts b/workspace/frontend/lib/i18n/messages/en-US.ts index e59fc8460..933a689ae 100644 --- a/workspace/frontend/lib/i18n/messages/en-US.ts +++ b/workspace/frontend/lib/i18n/messages/en-US.ts @@ -611,6 +611,26 @@ export const messages = { emptyDms: 'No agent conversations', }, + phaseGate: { + ownerMenuLabel: 'Clarification owner', + ownerHint: + 'This agent holds the floor. Others can be @mentioned for input, but they are kept in planning mode until you confirm the requirement.', + clarifyingWithOwner: 'Clarifying · @{name}', + clarifyingNeedsOwner: 'Clarifying · needs an owner', + clarifyingTitle: + 'The requirement is still being clarified — other agents can be consulted, but are kept in planning mode', + clarifyingOwnerlessTitle: + 'No agent owns this clarification — everyone is held in planning mode until you pick an owner', + turnOff: 'Turn the gate off', + confirm: 'Requirement confirmed', + confirmTitle: 'Release the gate — agents may start implementing', + building: 'Building', + clarifyFirst: 'Clarify first', + clarifyFirstTitle: + 'Hold the thread in clarification: only the owner keeps the floor until you confirm the requirement', + clarifyFirstMenuLabel: 'Clarify first — who owns the requirement?', + }, + newThread: { title: 'New Thread', descriptionMulti: 'Pick which agents join this conversation.', @@ -627,6 +647,19 @@ export const messages = { resumeNone: 'New conversation (no context)', start: 'Start Thread', resume: 'Resume Thread', + clarifyLabel: 'Clarify requirements before execution', + clarifyHint: + 'Recommended. One agent owns the requirement until you confirm it — the others can be asked for input, but are kept in planning mode.', + clarifyOwnerLabel: 'Who owns the requirement?', + clarifyOwnerPlaceholder: 'Select an agent…', + clarifyOwnerMaster: '{name} (workspace leader)', + clarifyOwnerRequired: 'Pick the agent that owns the requirement, or uncheck the option above.', + gateBlockedOffline: { + one: "{count} selected agent is offline and won't join, leaving too few to clarify with. Wait for them, pick another agent, or uncheck the option above to start without the gate.", + other: "{count} selected agents are offline and won't join, leaving too few to clarify with. Wait for them, pick another agent, or uncheck the option above to start without the gate.", + }, + gateBlockedTooFew: + 'Two or more online agents are needed to clarify. Pick another agent, or uncheck the option above.', }, landing: { diff --git a/workspace/frontend/lib/i18n/messages/zh-CN.ts b/workspace/frontend/lib/i18n/messages/zh-CN.ts index 858659143..b6c03a99d 100644 --- a/workspace/frontend/lib/i18n/messages/zh-CN.ts +++ b/workspace/frontend/lib/i18n/messages/zh-CN.ts @@ -594,6 +594,23 @@ export const messages: Messages = { emptyDms: '还没有与智能体的私聊', }, + phaseGate: { + ownerMenuLabel: '需求澄清负责人', + ownerHint: + '由这个智能体主导对话。其他智能体可以被 @ 征询意见,但在你确认需求之前,它们会保持在规划模式。', + clarifyingWithOwner: '澄清中 · @{name}', + clarifyingNeedsOwner: '澄清中 · 缺少负责人', + clarifyingTitle: '需求仍在澄清中 —— 其他智能体可以被征询,但会保持在规划模式', + clarifyingOwnerlessTitle: '这次澄清没有负责人 —— 在你指定负责人之前,所有智能体都停留在规划模式', + turnOff: '关闭澄清阶段', + confirm: '需求已确认', + confirmTitle: '解除限制 —— 智能体可以开始实现', + building: '实现中', + clarifyFirst: '先澄清需求', + clarifyFirstTitle: '让会话停在需求澄清阶段:在你确认需求之前,只有负责人主导对话', + clarifyFirstMenuLabel: '先澄清需求 —— 由谁负责?', + }, + newThread: { title: '新建会话', descriptionMulti: '选择要加入这个会话的智能体。', @@ -610,6 +627,18 @@ export const messages: Messages = { resumeNone: '全新对话(不带上下文)', start: '创建会话', resume: '继续会话', + clarifyLabel: '先澄清需求,再开始执行', + clarifyHint: + '推荐开启。在你确认需求之前,由一个智能体负责需求澄清;其他智能体可以被征询意见,但会保持在规划模式。', + clarifyOwnerLabel: '由谁负责需求?', + clarifyOwnerPlaceholder: '选择一个智能体…', + clarifyOwnerMaster: '{name}(工作区负责人)', + clarifyOwnerRequired: '请选择负责需求的智能体,或取消勾选上面的选项。', + gateBlockedOffline: { + one: '有 {count} 个已选中的智能体处于离线状态、不会加入,剩下的人数不足以进行需求澄清。可以等它上线、改选其他智能体,或取消勾选上面的选项以不带澄清阶段直接开始。', + other: '有 {count} 个已选中的智能体处于离线状态、不会加入,剩下的人数不足以进行需求澄清。可以等它们上线、改选其他智能体,或取消勾选上面的选项以不带澄清阶段直接开始。', + }, + gateBlockedTooFew: '需求澄清至少需要两个在线的智能体。请再选一个智能体,或取消勾选上面的选项。', }, landing: { diff --git a/workspace/frontend/lib/types.ts b/workspace/frontend/lib/types.ts index 3574f1fb4..c5ed4e39e 100644 --- a/workspace/frontend/lib/types.ts +++ b/workspace/frontend/lib/types.ts @@ -89,6 +89,10 @@ export interface WorkspaceSession { orchestrationMode: string; // Free-text collaboration plan used only in 'workflow' mode orchestrationInstruction: string | null; + // Requirement-clarification gate: 'open' | 'clarifying' | 'building' + phase: string; + // Agent holding the floor while clarifying (falls back to the master) + phaseOwner: string | null; createdAt: string | null; lastEventAt: number | null; // unix ms timestamp of last message } @@ -397,6 +401,8 @@ export interface NetworkChannel { master: string | null; orchestration_mode?: string; orchestration_instruction?: string | null; + phase?: string; + phase_owner?: string | null; participants: string[]; created_at: number | null; last_event_at: number | null; @@ -515,6 +521,8 @@ export function networkChannelToSession(ch: NetworkChannel, workspaceId: string) master: ch.master, orchestrationMode: ch.orchestration_mode || 'dynamic', orchestrationInstruction: ch.orchestration_instruction ?? null, + phase: ch.phase || 'open', + phaseOwner: ch.phase_owner ?? null, createdAt: ch.created_at ? new Date(ch.created_at).toISOString() : null, lastEventAt: ch.last_event_at, }; diff --git a/workspace/frontend/lib/workspace-context.tsx b/workspace/frontend/lib/workspace-context.tsx index aadea5a6a..c851fc3b2 100644 --- a/workspace/frontend/lib/workspace-context.tsx +++ b/workspace/frontend/lib/workspace-context.tsx @@ -146,13 +146,13 @@ interface WorkspaceContextValue { setSelectedFileId: (id: string | null) => void; setSelectedKnowledgeId: (id: string | null) => void; setCurrentFilePath: (path: string) => void; - createSession: (opts?: { title?: string; master?: string; participants?: string[]; resumeFrom?: string }) => Promise; + createSession: (opts?: { title?: string; master?: string; participants?: string[]; resumeFrom?: string; phase?: string; phaseOwner?: string }) => Promise; renameSession: (sessionId: string, title: string) => Promise; updateSession: (sessionId: string, updates: { starred?: boolean; status?: string }) => Promise; addParticipant: (sessionId: string, agentName: string) => Promise; removeParticipant: (sessionId: string, agentName: string) => Promise; setSessionMaster: (sessionId: string, agentName: string) => Promise; - setSessionOrchestration: (sessionId: string, updates: { mode?: string; instruction?: string | null }) => Promise; + setSessionOrchestration: (sessionId: string, updates: { mode?: string; instruction?: string | null; phase?: string; phaseOwner?: string | null }) => Promise; renameWorkspace: (name: string) => Promise; refreshWorkspace: () => Promise; refreshAgents: () => Promise; @@ -621,6 +621,8 @@ export function WorkspaceProvider({ master: remote.master, orchestrationMode: remote.orchestrationMode, orchestrationInstruction: remote.orchestrationInstruction, + phase: remote.phase, + phaseOwner: remote.phaseOwner, lastEventAt: remote.lastEventAt, createdAt: remote.createdAt || s.createdAt, status: remote.status, @@ -1285,7 +1287,7 @@ export function WorkspaceProvider({ return () => clearTimeout(timeout); }, [refreshDiscovery]); - const createSession = useCallback(async (opts?: { title?: string; master?: string; participants?: string[]; resumeFrom?: string }) => { + const createSession = useCallback(async (opts?: { title?: string; master?: string; participants?: string[]; resumeFrom?: string; phase?: string; phaseOwner?: string }) => { // Only set a channel leader when one is explicitly requested (e.g. the // single-agent DM path). The default "dynamic" orchestration mode needs no // leader, so threads created from the picker start with none — a leader can @@ -1298,6 +1300,8 @@ export function WorkspaceProvider({ master: masterAgent, participants, resumeFrom: opts?.resumeFrom, + phase: opts?.phase, + phaseOwner: opts?.phaseOwner, }); capture('thread_created', { participant_count: participants.length, has_resume: !!opts?.resumeFrom }); setSessions((prev) => [session, ...prev]); @@ -1347,7 +1351,7 @@ export function WorkspaceProvider({ const setSessionOrchestration = useCallback(async ( sessionId: string, - updates: { mode?: string; instruction?: string | null }, + updates: { mode?: string; instruction?: string | null; phase?: string; phaseOwner?: string | null }, ) => { // Optimistic: apply the mode/instruction locally, roll back on failure. // Snapshot the pre-update session inside the state updater so we read @@ -1363,6 +1367,8 @@ export function WorkspaceProvider({ orchestrationMode: updates.mode ?? s.orchestrationMode, orchestrationInstruction: updates.instruction !== undefined ? updates.instruction : s.orchestrationInstruction, + phase: updates.phase ?? s.phase, + phaseOwner: updates.phaseOwner !== undefined ? updates.phaseOwner : s.phaseOwner, }; }) ); @@ -1370,6 +1376,8 @@ export function WorkspaceProvider({ await workspaceApi.updateChannel(sessionId, { ...(updates.mode !== undefined && { orchestrationMode: updates.mode }), ...(updates.instruction !== undefined && { orchestrationInstruction: updates.instruction }), + ...(updates.phase !== undefined && { phase: updates.phase }), + ...(updates.phaseOwner !== undefined && { phaseOwner: updates.phaseOwner }), }); } catch { if (rollback.prev) {