mirror of
https://github.com/Ryder-Sun/Meldwork.git
synced 2026-11-07 06:56:03 -10:00
fix: render group replies in the order Agents were dispatched
No message carried a turn order. The timeline rebuilt one from round, phase, the Agent's current position in the group and the save time, and those disagreed: - live bubbles had numeric startedAt, which Date.parse turns into NaN, so every running Agent was pinned under the user message and the list reshuffled each time a reply was saved; - manual concurrent mode wrote failure notices before the barrier committed successful replies, so failures always rendered first; - round-less notices sorted after every round, the 'discussion' phase had no rank, and editing the group reordered history; - natural proposals were saved in completion order and peers read the transcript in that order, unlike what the user saw. Now every group topic message gets a per-topic turnSeq. Batch dispatchers (manual concurrent replies, natural proposal and discussion batches) reserve numbers in slot order before dispatching; other turns reserve at invocation start; failure and stop notices take their turn's number; other topic messages take the next number when written. The harness exposes the number on live Agent runs, so a live bubble and the reply that replaces it share one position. Numbers are derived from saved messages after a restart (no ledger or journal field changes). The renderer sorts group topics by turnSeq with one shared rule for the timeline and the trace panel; older history keeps its saved chronological order. A failed attempt whose retry was already saved no longer lingers as a ghost bubble, and peers read turns in dispatch order. Checks: new local-workspace-turn-order tests (slot order with a failing middle Agent and reversed completion; numbering continues after restart; none on direct chats), natural discussion, manual durability and write scheduling tests (63 pass on this tree); npm --prefix frontend test (330 pass, including a new conversationOrdering spec and a production-shaped live round test). Full desktop suite on the combined tree: 1529 of 1530, the remaining failure being the ACP timing test fixed in the next commit.
This commit is contained in:
1 parent
6522d357e0
commit
06e1e8772c
16 files changed
+359
-55
No files matched your search
@@ -149,7 +149,7 @@ class RunHarness {
|
||||
return next
|
||||
}
|
||||
|
||||
beginAgent(kind, round = 0, sourceMessageIds = []) {
|
||||
beginAgent(kind, round = 0, sourceMessageIds = [], options = {}) {
|
||||
const safeKind = cleanId(kind)
|
||||
if (!safeKind) throw new Error('RUN_HARNESS_AGENT_REQUIRED')
|
||||
if (!this.targetKinds.includes(safeKind)) {
|
||||
@@ -184,6 +184,10 @@ class RunHarness {
|
||||
seenSeqs: [],
|
||||
eventCursor: 0,
|
||||
context: {},
|
||||
// Per-topic dispatch order; the live bubble and the saved reply share it.
|
||||
...(Number.isSafeInteger(options?.turnSeq) && options.turnSeq > 0
|
||||
? { turnSeq: options.turnSeq }
|
||||
: {}),
|
||||
}
|
||||
this.agentRuns.push(run)
|
||||
while (this.agentRuns.length > this.maxAgentRuns) {
|
||||
@@ -328,6 +332,7 @@ class RunHarness {
|
||||
seenSeqs: [...run.seenSeqs],
|
||||
eventCursor: run.eventCursor,
|
||||
context: { ...run.context },
|
||||
...(run.turnSeq ? { turnSeq: run.turnSeq } : {}),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -517,6 +517,9 @@ class LocalWorkspaceAgentInvocation {
|
||||
this.scheduleRunCheckpoint = options.scheduleRunCheckpoint
|
||||
this.emitChanged = options.emitChanged
|
||||
this.promptFor = options.promptFor
|
||||
this.reserveTurnSeq = typeof options.reserveTurnSeq === 'function'
|
||||
? options.reserveTurnSeq
|
||||
: null
|
||||
this.agentLabel = typeof options.agentLabel === 'function'
|
||||
? options.agentLabel
|
||||
: kind => defaultAgentLabel(kind)
|
||||
@@ -551,6 +554,17 @@ class LocalWorkspaceAgentInvocation {
|
||||
}
|
||||
|
||||
async invoke(group, kind, mode, signal, threadRootId = '', context = {}) {
|
||||
// Group turns get a per-topic dispatch number before anything awaits, so
|
||||
// concurrent dispatches keep their call order. Batch dispatchers reserve
|
||||
// numbers themselves and pass them in.
|
||||
const requestedTurnSeq = Number(context.turnSeq)
|
||||
const turnSeq = Number.isSafeInteger(requestedTurnSeq) && requestedTurnSeq > 0
|
||||
? requestedTurnSeq
|
||||
: (group.conversationType !== 'direct' && threadRootId
|
||||
&& typeof this.reserveTurnSeq === 'function'
|
||||
? this.reserveTurnSeq(group.id, threadRootId)
|
||||
: null)
|
||||
if (turnSeq) context = { ...context, turnSeq }
|
||||
const activeRun = this.activeRuns.get(group.id)
|
||||
const taskId = cleanText(activeRun?.taskId || context.taskId || threadRootId, 120)
|
||||
if (!taskId) throw new Error('LOCAL_RUN_TASK_INVALID')
|
||||
@@ -877,7 +891,9 @@ class LocalWorkspaceAgentInvocation {
|
||||
if (!isolated && !frozen) assertRequiredContextFits(packedContext)
|
||||
let deliveredContext = deliveredPackedContext(packedContext, promptMode)
|
||||
const harness = this.ensureRunHarness(group, activeRun, threadRootId)
|
||||
const harnessRun = harness?.beginAgent(kind, round, deliveredContext.sourceMessageIds)
|
||||
const harnessRun = harness?.beginAgent(kind, round, deliveredContext.sourceMessageIds, {
|
||||
turnSeq: context.turnSeq,
|
||||
})
|
||||
let deliveryContext = {
|
||||
contextPackId: invocationContext?.contextPackId || activeRun?.contextPackId || '',
|
||||
deliveryRecordIds: [],
|
||||
@@ -1936,6 +1952,7 @@ class LocalWorkspaceAgentInvocation {
|
||||
attachments,
|
||||
trace,
|
||||
responseVersionRootId,
|
||||
...(context.turnSeq ? { turnSeq: context.turnSeq } : {}),
|
||||
}
|
||||
const pendingMessage = internal
|
||||
? null
|
||||
@@ -2037,6 +2054,14 @@ class LocalWorkspaceAgentInvocation {
|
||||
configurable: true,
|
||||
})
|
||||
}
|
||||
// Failure and stop notices take the turn's slot in the timeline.
|
||||
if (context.turnSeq && error && (typeof error === 'object' || typeof error === 'function')) {
|
||||
Object.defineProperty(error, 'turnSeq', {
|
||||
value: context.turnSeq,
|
||||
enumerable: false,
|
||||
configurable: true,
|
||||
})
|
||||
}
|
||||
throw attachInvocationFailure(error, {
|
||||
outcomeCertainty: terminalFailureKnown
|
||||
? 'known_failed'
|
||||
|
||||
@@ -94,6 +94,9 @@ class LocalWorkspaceAutoRunner {
|
||||
this.recordAgentFailure = options.recordAgentFailure
|
||||
this.recordAgentInterruption = options.recordAgentInterruption
|
||||
this.addMessage = options.addMessage
|
||||
this.reserveTurnSeqs = typeof options.reserveTurnSeqs === 'function'
|
||||
? options.reserveTurnSeqs
|
||||
: () => []
|
||||
this.commitV4AgentMessage = typeof options.commitV4AgentMessage === 'function'
|
||||
? options.commitV4AgentMessage
|
||||
: null
|
||||
@@ -2508,7 +2511,18 @@ class LocalWorkspaceAutoRunner {
|
||||
&& (!options.controller || this.v4NaturalMessageMatchesBinding(
|
||||
message, options.controller,
|
||||
))
|
||||
)).map(message => ({
|
||||
)).map((message, index) => ({ message, index }))
|
||||
// Peers read turns in dispatch order, the same order the user sees,
|
||||
// instead of the order replies happened to be saved in.
|
||||
.sort((left, right) => {
|
||||
const leftTurn = Number.isSafeInteger(left.message.turnSeq) ? left.message.turnSeq : 0
|
||||
const rightTurn = Number.isSafeInteger(right.message.turnSeq) ? right.message.turnSeq : 0
|
||||
return (leftTurn ? 1 : 0) - (rightTurn ? 1 : 0)
|
||||
|| leftTurn - rightTurn
|
||||
|| left.index - right.index
|
||||
})
|
||||
.map(({ message }) => message)
|
||||
.map(message => ({
|
||||
...this.v4NaturalMessageContent(message),
|
||||
id: message.id,
|
||||
agentKind: message.agentKind,
|
||||
@@ -3902,7 +3916,8 @@ class LocalWorkspaceAutoRunner {
|
||||
this.emitChanged()
|
||||
|
||||
if (phase === 'proposal') {
|
||||
const dispatches = activeKinds.map(kind => {
|
||||
const proposalTurns = this.reserveTurnSeqs(group.id, threadRootId, activeKinds.length)
|
||||
const dispatches = activeKinds.map((kind, kindIndex) => {
|
||||
const slot = phaseSlots.find(candidate => candidate.agentKind === kind)
|
||||
const prompt = this.v4PhasePrompt(
|
||||
group, kind, phase, snapshot, receiptRecords, 'participant', {
|
||||
@@ -3921,6 +3936,7 @@ class LocalWorkspaceAutoRunner {
|
||||
batchId,
|
||||
slotId: slot.slotId,
|
||||
snapshotHash,
|
||||
...(proposalTurns[kindIndex] ? { turnSeq: proposalTurns[kindIndex] } : {}),
|
||||
...(sessionPolicy ? { sessionPolicy } : {}),
|
||||
promptOverride: prompt,
|
||||
contextPackId: controller.contextPackId,
|
||||
@@ -3988,6 +4004,10 @@ class LocalWorkspaceAutoRunner {
|
||||
workAdmissionTail = admission.catch(() => { workAdmissionClosed = true })
|
||||
return admission
|
||||
}
|
||||
// Batched peers keep dispatch order in the timeline whatever order they
|
||||
// finish in.
|
||||
const phaseTurns = this.reserveTurnSeqs(group.id, threadRootId, activeKinds.length)
|
||||
const turnSeqByKind = new Map(activeKinds.map((kind, index) => [kind, phaseTurns[index]]))
|
||||
const settled = await Promise.all(activeKinds.map(async kind => {
|
||||
const slot = phaseSlots.find(candidate => candidate.agentKind === kind)
|
||||
const challengeBinding = phase === 'challenge'
|
||||
@@ -4049,6 +4069,7 @@ class LocalWorkspaceAutoRunner {
|
||||
batchId,
|
||||
slotId: slot.slotId,
|
||||
snapshotHash,
|
||||
...(turnSeqByKind.get(kind) ? { turnSeq: turnSeqByKind.get(kind) } : {}),
|
||||
...(sessionPolicy ? { sessionPolicy } : {}),
|
||||
promptOverride: prompt,
|
||||
contextPackId: controller.contextPackId,
|
||||
|
||||
@@ -16,6 +16,7 @@ const {
|
||||
|
||||
class LocalWorkspaceConversations {
|
||||
constructor(options) {
|
||||
this.reserveTurnSeq = typeof options.reserveTurnSeq === 'function' ? options.reserveTurnSeq : null
|
||||
this.state = options.state
|
||||
this.detectedAgents = options.detectedAgents
|
||||
this.save = options.save
|
||||
@@ -239,6 +240,16 @@ class LocalWorkspaceConversations {
|
||||
createdAt: this.now(),
|
||||
}
|
||||
if (threadRootId) message.threadRootId = threadRootId
|
||||
// Group topics order by dispatch turn. Replies and failure notices carry
|
||||
// the number reserved at dispatch; any other topic message (system
|
||||
// notices, follow-up input) takes the next number when it is written.
|
||||
const requestedTurnSeq = Number(metadata.turnSeq)
|
||||
const turnSeq = Number.isSafeInteger(requestedTurnSeq) && requestedTurnSeq > 0
|
||||
? requestedTurnSeq
|
||||
: (threadRootId && group.conversationType !== 'direct' && this.reserveTurnSeq
|
||||
? this.reserveTurnSeq(groupId, threadRootId)
|
||||
: null)
|
||||
if (turnSeq) message.turnSeq = turnSeq
|
||||
if (systemKey) {
|
||||
message.system = {
|
||||
key: systemKey,
|
||||
|
||||
@@ -215,8 +215,10 @@ function normalizeLoadedMessage(input) {
|
||||
const agentKind = isSupportedAgentKind(requestedAgentKind) ? requestedAgentKind : ''
|
||||
if (!id || !groupId || !role || (role === 'agent' && !agentKind)) return null
|
||||
const systemKey = role === 'system' ? cleanInline(input.system?.key, 100) : ''
|
||||
const turnSeq = Number(input.turnSeq)
|
||||
|
||||
const message = {
|
||||
...(Number.isSafeInteger(turnSeq) && turnSeq > 0 ? { turnSeq } : {}),
|
||||
id,
|
||||
groupId,
|
||||
role,
|
||||
|
||||
@@ -108,6 +108,9 @@ class LocalWorkspaceMessageSubmission {
|
||||
this.bindRunTask = options.bindRunTask
|
||||
this.releasePreparation = options.releasePreparation
|
||||
this.addMessage = options.addMessage
|
||||
this.reserveTurnSeqs = typeof options.reserveTurnSeqs === 'function'
|
||||
? options.reserveTurnSeqs
|
||||
: () => []
|
||||
this.rollbackAddedMessage = options.rollbackAddedMessage
|
||||
this.startAutoRunner = options.startAutoRunner
|
||||
this.beginRun = options.beginRun
|
||||
@@ -545,6 +548,7 @@ class LocalWorkspaceMessageSubmission {
|
||||
...(pendingMessage.metadata?.responseVersionRootId
|
||||
? { responseVersionRootId: pendingMessage.metadata.responseVersionRootId } : {}),
|
||||
...(trace ? { trace } : {}),
|
||||
...(pendingMessage.metadata?.turnSeq ? { turnSeq: pendingMessage.metadata.turnSeq } : {}),
|
||||
},
|
||||
}))
|
||||
const contentRef = this.contentBlobStore.put(canonicalJson(body), {
|
||||
@@ -887,6 +891,10 @@ class LocalWorkspaceMessageSubmission {
|
||||
snapshot,
|
||||
skillHints: skillHintsByKind.get(kind) || [],
|
||||
})]))
|
||||
// Concurrent replies keep slot order in the timeline whatever order they
|
||||
// finish in; failure notices take their slot's number too.
|
||||
const reservedTurns = this.reserveTurnSeqs(group.id, threadRootId, activeKinds.length)
|
||||
const turnSeqByKind = new Map(activeKinds.map((kind, index) => [kind, reservedTurns[index]]))
|
||||
const pending = await Promise.all(activeKinds.map(async (kind) => {
|
||||
const index = targetKinds.indexOf(kind)
|
||||
const slot = slots[index]
|
||||
@@ -923,6 +931,7 @@ class LocalWorkspaceMessageSubmission {
|
||||
slotId: slot.slotId,
|
||||
snapshotHash: frozenSnapshotHash,
|
||||
sessionPolicy: 'frozen',
|
||||
...(turnSeqByKind.get(kind) ? { turnSeq: turnSeqByKind.get(kind) } : {}),
|
||||
promptOverride: prompts.get(kind),
|
||||
contextPackId: controller.contextPackId,
|
||||
snapshotSourceMessageIds,
|
||||
|
||||
@@ -9,6 +9,11 @@ const {
|
||||
terminalStatusPrefix,
|
||||
} = require('./local-workspace-inputs.cjs')
|
||||
|
||||
function turnSeqMetadata(error) {
|
||||
const turnSeq = Number(error?.turnSeq)
|
||||
return Number.isSafeInteger(turnSeq) && turnSeq > 0 ? { turnSeq } : {}
|
||||
}
|
||||
|
||||
class LocalWorkspaceRunMessages {
|
||||
constructor(options) {
|
||||
this.state = options.state
|
||||
@@ -73,7 +78,7 @@ class LocalWorkspaceRunMessages {
|
||||
kind,
|
||||
threadRootId,
|
||||
system,
|
||||
{ trace, responseVersionRootId },
|
||||
{ trace, responseVersionRootId, ...turnSeqMetadata(error) },
|
||||
)
|
||||
}
|
||||
return { label, reason }
|
||||
@@ -114,7 +119,7 @@ class LocalWorkspaceRunMessages {
|
||||
key: interrupted ? 'system.agentInterrupted' : 'system.agentStopped',
|
||||
params: { agent: label },
|
||||
},
|
||||
{ trace, responseVersionRootId },
|
||||
{ trace, responseVersionRootId, ...turnSeqMetadata(error) },
|
||||
)
|
||||
return trace
|
||||
}
|
||||
|
||||
@@ -192,6 +192,10 @@ class LocalWorkspace extends EventEmitter {
|
||||
this.detectedAgents = []
|
||||
this.preparingRuns = new Map()
|
||||
this.activeRuns = new Map()
|
||||
// Highest turn number handed out per group topic in this process. Saved
|
||||
// messages are the durable source; reservations only cover turns that are
|
||||
// dispatched but not written yet.
|
||||
this.turnSeqReservations = new Map()
|
||||
this.pendingV4ManualRecoveries = new Map()
|
||||
this.pendingV4AutoRecoveries = new Map()
|
||||
this.runCheckpointTimers = new Map()
|
||||
@@ -247,6 +251,7 @@ class LocalWorkspace extends EventEmitter {
|
||||
terminalRetrySleep: options.terminalRetrySleep,
|
||||
})
|
||||
this.conversations = new LocalWorkspaceConversations({
|
||||
reserveTurnSeq: (groupId, threadRootId) => this.reserveTurnSeqs(groupId, threadRootId, 1)[0],
|
||||
state: () => this.state,
|
||||
detectedAgents: () => this.detectedAgents,
|
||||
save: () => this.save(),
|
||||
@@ -266,6 +271,7 @@ class LocalWorkspace extends EventEmitter {
|
||||
addMessage: (...args) => this.addMessage(...args),
|
||||
})
|
||||
this.agentInvocation = new LocalWorkspaceAgentInvocation({
|
||||
reserveTurnSeq: (groupId, threadRootId) => this.reserveTurnSeqs(groupId, threadRootId, 1)[0],
|
||||
state: () => this.state,
|
||||
agentLabel: kind => this.agentLabel(kind),
|
||||
detectedAgents: () => this.detectedAgents,
|
||||
@@ -311,6 +317,7 @@ class LocalWorkspace extends EventEmitter {
|
||||
attachmentSupport: (...args) => this.attachmentSupportFn(...args),
|
||||
})
|
||||
this.autoRunner = new LocalWorkspaceAutoRunner({
|
||||
reserveTurnSeqs: (...args) => this.reserveTurnSeqs(...args),
|
||||
state: () => this.state,
|
||||
canWriteWorkspace: kind => agentRuntimeCapabilities(kind, {
|
||||
agent: this.detectedAgents.find(agent => agent.kind === kind),
|
||||
@@ -352,6 +359,7 @@ class LocalWorkspace extends EventEmitter {
|
||||
retrySleep: options.retrySleep,
|
||||
})
|
||||
this.messageSubmission = new LocalWorkspaceMessageSubmission({
|
||||
reserveTurnSeqs: (...args) => this.reserveTurnSeqs(...args),
|
||||
state: () => this.state,
|
||||
detectedAgents: () => this.detectedAgents,
|
||||
isShuttingDown: () => this.shuttingDown,
|
||||
@@ -2430,6 +2438,22 @@ class LocalWorkspace extends EventEmitter {
|
||||
)
|
||||
}
|
||||
|
||||
// Reserves `count` consecutive dispatch-turn numbers for one group topic.
|
||||
reserveTurnSeqs(groupId, threadRootId, count = 1) {
|
||||
const amount = Math.max(0, Math.min(64, Math.floor(Number(count) || 0)))
|
||||
if (!groupId || !threadRootId || !amount) return []
|
||||
const key = `${groupId}\u0000${threadRootId}`
|
||||
let cursor = this.turnSeqReservations.get(key) || 0
|
||||
for (const message of this.state.messages) {
|
||||
if (message.groupId === groupId && message.threadRootId === threadRootId
|
||||
&& Number.isSafeInteger(message.turnSeq) && message.turnSeq > cursor) {
|
||||
cursor = message.turnSeq
|
||||
}
|
||||
}
|
||||
this.turnSeqReservations.set(key, cursor + amount)
|
||||
return Array.from({ length: amount }, (_, index) => cursor + index + 1)
|
||||
}
|
||||
|
||||
commitV4AgentMessage(input = {}) {
|
||||
const messageId = String(input.messageId || '')
|
||||
const existing = this.state.messages.find(message => message.id === messageId)
|
||||
@@ -2459,6 +2483,11 @@ class LocalWorkspace extends EventEmitter {
|
||||
...(input.metadata?.responseVersionRootId
|
||||
? { responseVersionRootId: input.metadata.responseVersionRootId } : {}),
|
||||
...(input.metadata?.trace ? { trace: input.metadata.trace } : {}),
|
||||
...(input.metadata?.turnSeq
|
||||
? { turnSeq: input.metadata.turnSeq }
|
||||
: (input.threadRootId && group.conversationType !== 'direct'
|
||||
? { turnSeq: this.reserveTurnSeqs(group.id, input.threadRootId, 1)[0] }
|
||||
: {})),
|
||||
})
|
||||
if (!message || message.id !== messageId) throw new Error('LOCAL_RUN_COMMIT_INVALID')
|
||||
const previousMessages = [...this.state.messages]
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
const test = require('node:test')
|
||||
const assert = require('node:assert/strict')
|
||||
const fs = require('node:fs')
|
||||
const { LocalWorkspace } = require('../../src/workspace/local-workspace.cjs')
|
||||
const { agentRuntimeError } = require('../../src/agents/agent-runtime-contract.cjs')
|
||||
const { fixture } = require('../support/local-workspace-test-helpers.cjs')
|
||||
|
||||
function topicMessages(workspace, rootId) {
|
||||
return workspace.snapshot().messages.filter(message => message.threadRootId === rootId)
|
||||
}
|
||||
|
||||
test('concurrent replies and failure notices keep slot order whatever order they finish in', async (t) => {
|
||||
const { directory, options } = fixture()
|
||||
t.after(() => fs.rmSync(directory, { recursive: true, force: true }))
|
||||
options.retrySleep = async () => {}
|
||||
options.runAgent = async (agent, _prompt, _cwd, runOptions) => {
|
||||
// Codex finishes last, Hermes fails first, WorkBuddy finishes in between.
|
||||
if (agent.kind === 'codex') {
|
||||
await new Promise(resolve => setTimeout(resolve, 40))
|
||||
return { text: 'codex result', sessionRef: runOptions.sessionRef || 'codex-session' }
|
||||
}
|
||||
if (agent.kind === 'hermes') throw agentRuntimeError('LOCAL_AGENT_REFUSED', 'refused')
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
return { text: 'workbuddy result', sessionRef: runOptions.sessionRef || 'workbuddy-session' }
|
||||
}
|
||||
const workspace = new LocalWorkspace(options)
|
||||
await workspace.refreshAgents()
|
||||
const group = workspace.createGroup({
|
||||
name: 'Ordered batch', agentKinds: ['codex', 'hermes', 'workbuddy'],
|
||||
workdir: directory, allowWrite: false,
|
||||
})
|
||||
|
||||
await workspace.sendMessage({
|
||||
groupId: group.id, text: 'Answer independently.',
|
||||
protocol: 'v4', mode: 'manual', targetKinds: group.agentKinds,
|
||||
})
|
||||
|
||||
const root = workspace.snapshot().messages.find(message => message.role === 'user')
|
||||
const topic = topicMessages(workspace, root.id)
|
||||
const byTurn = [...topic].sort((left, right) => left.turnSeq - right.turnSeq)
|
||||
assert.deepEqual(byTurn.map(message => [message.role, message.agentKind]), [
|
||||
['agent', 'codex'],
|
||||
['system', 'hermes'],
|
||||
['agent', 'workbuddy'],
|
||||
])
|
||||
assert.equal(topic.every(message => Number.isSafeInteger(message.turnSeq)), true)
|
||||
assert.equal(new Set(topic.map(message => message.turnSeq)).size, topic.length)
|
||||
})
|
||||
|
||||
test('turn numbers continue after a restart and stay off direct chats', async (t) => {
|
||||
const { directory, options } = fixture()
|
||||
t.after(() => fs.rmSync(directory, { recursive: true, force: true }))
|
||||
options.runAgent = async (agent, _prompt, _cwd, runOptions) => ({
|
||||
text: `${agent.kind} reply`, sessionRef: runOptions.sessionRef || `${agent.kind}-session`,
|
||||
})
|
||||
const workspace = new LocalWorkspace(options)
|
||||
await workspace.refreshAgents()
|
||||
const group = workspace.createGroup({
|
||||
name: 'Numbered topic', agentKinds: ['codex', 'hermes'], workdir: directory, allowWrite: false,
|
||||
})
|
||||
await workspace.sendMessage({
|
||||
groupId: group.id, text: 'First turn', protocol: 'v4', mode: 'manual',
|
||||
targetKinds: group.agentKinds,
|
||||
})
|
||||
const root = workspace.snapshot().messages.find(message => message.role === 'user')
|
||||
const before = Math.max(...topicMessages(workspace, root.id).map(message => message.turnSeq))
|
||||
|
||||
const restarted = new LocalWorkspace(options)
|
||||
await restarted.refreshAgents()
|
||||
const notice = restarted.addMessage(group.id, 'system', 'Follow-up notice', '', root.id)
|
||||
assert.equal(notice.turnSeq, before + 1)
|
||||
|
||||
const direct = restarted.createGroup({
|
||||
name: 'Codex', agentKinds: ['codex'], workdir: directory,
|
||||
conversationType: 'direct', directAgentKind: 'codex',
|
||||
})
|
||||
await restarted.sendMessage({ groupId: direct.id, text: 'Hello' })
|
||||
assert.equal(restarted.snapshot().messages
|
||||
.filter(message => message.groupId === direct.id)
|
||||
.some(message => message.turnSeq != null), false)
|
||||
})
|
||||
@@ -166,7 +166,7 @@ describe('Meldwork workbench', () => {
|
||||
wrapper.unmount()
|
||||
})
|
||||
|
||||
it('keeps live Round 2 Agent replies after Round 1 in sequential execution order', async () => {
|
||||
it('keeps live Round 2 Agent replies after Round 1 in dispatch turn order', async () => {
|
||||
const { wrapper } = await mountApp(({ state }) => {
|
||||
state.groups.push({
|
||||
id: 'group-live-round-order',
|
||||
@@ -194,6 +194,7 @@ describe('Meldwork workbench', () => {
|
||||
agentKind: 'codex',
|
||||
content: 'Codex proposal',
|
||||
threadRootId: 'live-round-root',
|
||||
turnSeq: 1,
|
||||
trace: {
|
||||
runId: 'run-live-round-order',
|
||||
agentRunId: 'codex-round-1',
|
||||
@@ -211,6 +212,7 @@ describe('Meldwork workbench', () => {
|
||||
agentKind: 'hermes',
|
||||
content: 'Hermes proposal',
|
||||
threadRootId: 'live-round-root',
|
||||
turnSeq: 2,
|
||||
trace: {
|
||||
runId: 'run-live-round-order',
|
||||
agentRunId: 'hermes-round-1',
|
||||
@@ -235,23 +237,23 @@ describe('Meldwork workbench', () => {
|
||||
agentRunId: 'codex-round-2',
|
||||
kind: 'codex',
|
||||
round: 2,
|
||||
executionSequence: 1,
|
||||
turnSeq: 3,
|
||||
status: 'completed',
|
||||
output: 'Codex challenge',
|
||||
events: [],
|
||||
context: { phase: 'challenge' },
|
||||
startedAt: '2026-07-29T07:59:00Z',
|
||||
// Production live runs carry numeric timestamps; the order must
|
||||
// come from the dispatch turn, not from these (reversed) times.
|
||||
startedAt: Date.parse('2026-07-29T07:59:00Z'),
|
||||
},
|
||||
{
|
||||
agentRunId: 'hermes-round-2',
|
||||
kind: 'hermes',
|
||||
round: 2,
|
||||
executionSequence: 2,
|
||||
turnSeq: 4,
|
||||
status: 'running',
|
||||
output: 'Hermes challenge',
|
||||
events: [],
|
||||
context: { phase: 'challenge' },
|
||||
startedAt: '2026-07-29T07:58:00Z',
|
||||
startedAt: Date.parse('2026-07-29T07:58:00Z'),
|
||||
},
|
||||
],
|
||||
}]
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
compareSortKeys,
|
||||
compareTraceItems,
|
||||
timelineSortKey,
|
||||
timestampValue,
|
||||
} from '../../conversationOrdering.js'
|
||||
|
||||
function order(messages, topicIndex = 0) {
|
||||
return messages
|
||||
.map((message, index) => ({ message, key: timelineSortKey(message, { topicIndex, index }) }))
|
||||
.sort((left, right) => compareSortKeys(left.key, right.key))
|
||||
.map(item => item.message.id)
|
||||
}
|
||||
|
||||
describe('conversation ordering', () => {
|
||||
it('parses numeric and ISO timestamps instead of collapsing numbers to zero', () => {
|
||||
expect(timestampValue(1760500000000)).toBe(1760500000000)
|
||||
expect(timestampValue('2026-10-15T08:00:00.000Z')).toBe(Date.parse('2026-10-15T08:00:00.000Z'))
|
||||
expect(timestampValue('not a date')).toBe(0)
|
||||
})
|
||||
|
||||
it('orders a topic by dispatch turn regardless of save time', () => {
|
||||
const root = { id: 'root', role: 'user' }
|
||||
const failure = { id: 'failure', role: 'system', threadRootId: 'root', turnSeq: 2, createdAt: 10 }
|
||||
const first = { id: 'first', role: 'agent', threadRootId: 'root', turnSeq: 1, createdAt: 30 }
|
||||
const third = { id: 'third', role: 'agent', threadRootId: 'root', turnSeq: 3, createdAt: 20 }
|
||||
const limit = { id: 'limit', role: 'system', threadRootId: 'root', turnSeq: 4, createdAt: 15 }
|
||||
expect(order([limit, third, failure, first, root])).toEqual(['root', 'first', 'failure', 'third', 'limit'])
|
||||
})
|
||||
|
||||
it('keeps a live bubble in the slot of the reply that will replace it', () => {
|
||||
const root = { id: 'root', role: 'user' }
|
||||
const saved = { id: 'saved', role: 'agent', threadRootId: 'root', turnSeq: 2, createdAt: 50 }
|
||||
const live = {
|
||||
id: 'live', role: 'agent', threadRootId: 'root', provisional: true,
|
||||
createdAt: 99, liveAgentRun: { turnSeq: 1 },
|
||||
}
|
||||
expect(order([saved, live, root])).toEqual(['root', 'live', 'saved'])
|
||||
})
|
||||
|
||||
it('keeps legacy history chronological and before newly numbered turns', () => {
|
||||
const root = { id: 'root', role: 'user' }
|
||||
const oldReply = { id: 'old', role: 'agent', threadRootId: 'root', createdAt: '2026-01-01T00:00:02Z' }
|
||||
const oldNotice = { id: 'notice', role: 'system', threadRootId: 'root', createdAt: '2026-01-01T00:00:01Z' }
|
||||
const numbered = { id: 'numbered', role: 'agent', threadRootId: 'root', turnSeq: 1, createdAt: 0 }
|
||||
const unnumberedLive = { id: 'live', role: 'agent', threadRootId: 'root', provisional: true, createdAt: 5 }
|
||||
expect(order([unnumberedLive, numbered, oldReply, oldNotice, root]))
|
||||
.toEqual(['root', 'notice', 'old', 'numbered', 'live'])
|
||||
})
|
||||
|
||||
it('orders trace items by round, then turn, then the run target order', () => {
|
||||
const targets = new Map([['hermes', 0], ['codex', 1]])
|
||||
const items = [
|
||||
{ agentKind: 'hermes', round: 2, turnSeq: 4 },
|
||||
{ agentKind: 'codex', round: 2, turnSeq: 3 },
|
||||
{ agentKind: 'codex', round: 1 },
|
||||
{ agentKind: 'hermes', round: 1 },
|
||||
]
|
||||
expect([...items].sort((left, right) => compareTraceItems(left, right, targets))
|
||||
.map(item => `${item.round}:${item.agentKind}`))
|
||||
.toEqual(['1:hermes', '1:codex', '2:codex', '2:hermes'])
|
||||
})
|
||||
})
|
||||
@@ -1,4 +1,10 @@
|
||||
import { computed, ref } from 'vue'
|
||||
import {
|
||||
compareSortKeys,
|
||||
positiveSequence,
|
||||
timelineSortKey,
|
||||
timestampValue,
|
||||
} from '../conversationOrdering.js'
|
||||
import { agentLabel } from '../catalog.js'
|
||||
import {
|
||||
durableRunTurnStatus,
|
||||
@@ -197,8 +203,19 @@ export function useConversationTimeline({
|
||||
.map(message => message?.trace?.agentRunId)
|
||||
.filter(Boolean))
|
||||
const rootId = run.threadRootId || topLevelUserMessages.value.at(-1)?.id || ''
|
||||
// While a retry is still live, earlier attempts render as response
|
||||
// versions of the same turn. Once the retry's reply is saved, a failed
|
||||
// earlier attempt has no message of its own and would linger as a ghost
|
||||
// bubble next to the real reply, so it is dropped.
|
||||
const supersededAttempt = agent => ['failed', 'timeout', 'stopped', 'interrupted'].includes(agent.status)
|
||||
&& activeRunAgentRuns.value.some(other => (
|
||||
other !== agent && other.kind === agent.kind
|
||||
&& durableAgentRunIds.has(other.agentRunId)
|
||||
&& timestampValue(other.startedAt) > timestampValue(agent.startedAt)
|
||||
))
|
||||
return activeRunAgentRuns.value
|
||||
.filter(agent => !durableAgentRunIds.has(agent.agentRunId))
|
||||
.filter(agent => !supersededAttempt(agent))
|
||||
.filter(agent => group.conversationType === 'direct' || agent.output || (agent.events || []).some(event => event.type !== 'answer_delta'))
|
||||
.map(agent => ({
|
||||
id: `run-message-${agent.agentRunId}`,
|
||||
@@ -207,6 +224,7 @@ export function useConversationTimeline({
|
||||
agentKind: agent.kind,
|
||||
content: agent.output || '',
|
||||
createdAt: agent.startedAt || Date.now(),
|
||||
...(positiveSequence(agent.turnSeq) ? { turnSeq: positiveSequence(agent.turnSeq) } : {}),
|
||||
threadRootId: rootId,
|
||||
provisional: true,
|
||||
traceRunId: run.runId,
|
||||
@@ -509,49 +527,20 @@ export function useConversationTimeline({
|
||||
return !rootId || isTopicExpanded(rootId)
|
||||
})
|
||||
if (activeGroup.value?.conversationType === 'direct') return messages
|
||||
const agentOrder = new Map((activeGroup.value?.agentKinds || []).map((kind, index) => [kind, index]))
|
||||
const phaseOrder = new Map([
|
||||
['proposal', 0], ['challenge', 1], ['coordination', 2], ['work', 3],
|
||||
['synthesis', 4], ['verification', 5], ['commit', 6],
|
||||
])
|
||||
const messageTime = message => {
|
||||
const parsed = Date.parse(message?.createdAt || '')
|
||||
return Number.isFinite(parsed) ? parsed : 0
|
||||
}
|
||||
const topicOrder = new Map(topLevelUserMessages.value.map((message, index) => [message.id, index]))
|
||||
const unanchoredTopicOrder = topLevelUserMessages.value.length + 1
|
||||
const rank = (message, index) => {
|
||||
const topicIndexFor = (message) => {
|
||||
const rootId = message.role === 'user' && !message.threadRootId
|
||||
? message.id
|
||||
: messageThreadRootId(message)
|
||||
const topicIndex = rootId && topicOrder.has(rootId)
|
||||
? topicOrder.get(rootId)
|
||||
: unanchoredTopicOrder
|
||||
if (message.role === 'user' && !message.threadRootId) return [topicIndex, 0, 0, 0, index]
|
||||
const trace = message.trace || {}
|
||||
const liveAgentRun = message.liveAgentRun || {}
|
||||
const round = Number(trace.round ?? liveAgentRun.round)
|
||||
const executionSequence = Number(trace.executionSequence ?? liveAgentRun.executionSequence)
|
||||
const phase = phaseOrder.get(String(
|
||||
trace.phase || message.tracePhase || liveAgentRun.context?.phase || '',
|
||||
).toLowerCase()) ?? 9
|
||||
const kind = agentOrder.get(message.agentKind) ?? 99
|
||||
if (Number.isInteger(round) && round > 0) {
|
||||
if (Number.isInteger(executionSequence) && executionSequence > 0) {
|
||||
return [topicIndex, 1, round, 0, executionSequence, index]
|
||||
}
|
||||
return [topicIndex, 1, round, 1, phase * 100 + kind, index]
|
||||
}
|
||||
return [topicIndex, 1, messageTime(message), kind, index]
|
||||
return rootId && topicOrder.has(rootId) ? topicOrder.get(rootId) : unanchoredTopicOrder
|
||||
}
|
||||
return messages
|
||||
.map((message, index) => ({ message, key: rank(message, index) }))
|
||||
.sort((left, right) => {
|
||||
for (let index = 0; index < left.key.length; index += 1) {
|
||||
if (left.key[index] !== right.key[index]) return left.key[index] - right.key[index]
|
||||
}
|
||||
return 0
|
||||
})
|
||||
.map((message, index) => ({
|
||||
message,
|
||||
key: timelineSortKey(message, { topicIndex: topicIndexFor(message), index }),
|
||||
}))
|
||||
.sort((left, right) => compareSortKeys(left.key, right.key))
|
||||
.map(item => item.message)
|
||||
})
|
||||
const conversationEmptyVisible = computed(() => (
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { computed } from 'vue'
|
||||
import { compareTraceItems } from '../conversationOrdering.js'
|
||||
import { agentLabel } from '../catalog.js'
|
||||
import { normalizeOrchestration } from '../desktop-normalization.js'
|
||||
import {
|
||||
@@ -69,6 +70,7 @@ export function useConversationTraceRuns({
|
||||
threadRootId: activeRun.value?.threadRootId || '',
|
||||
createdAt: agent.startedAt,
|
||||
startedAt: agent.startedAt,
|
||||
turnSeq: agent.turnSeq,
|
||||
live: true,
|
||||
})
|
||||
}
|
||||
@@ -94,6 +96,7 @@ export function useConversationTraceRuns({
|
||||
messageId: message.id,
|
||||
threadRootId: messageThreadRootId(message),
|
||||
createdAt: message.createdAt,
|
||||
turnSeq: message.turnSeq,
|
||||
live: false,
|
||||
}
|
||||
const live = byRunAgentId.get(key)
|
||||
@@ -112,17 +115,13 @@ export function useConversationTraceRuns({
|
||||
threadRootId: live.threadRootId || durable.threadRootId,
|
||||
createdAt: live.createdAt || durable.createdAt,
|
||||
startedAt: live.startedAt || durable.startedAt,
|
||||
turnSeq: durable.turnSeq || live.turnSeq,
|
||||
live: true,
|
||||
} : durable)
|
||||
}
|
||||
const targetOrder = new Map((activeRun.value?.targetKinds
|
||||
|| activeGroup.value?.agentKinds || []).map((kind, index) => [kind, index]))
|
||||
return [...byRunAgentId.values()].sort((left, right) => (
|
||||
(Number(left.round) || 0) - (Number(right.round) || 0)
|
||||
|| (targetOrder.get(left.agentKind) ?? 99) - (targetOrder.get(right.agentKind) ?? 99)
|
||||
|| (Date.parse(left.createdAt || left.startedAt || '') || 0)
|
||||
- (Date.parse(right.createdAt || right.startedAt || '') || 0)
|
||||
))
|
||||
return [...byRunAgentId.values()].sort((left, right) => compareTraceItems(left, right, targetOrder))
|
||||
})
|
||||
|
||||
const historicalGroupRun = computed(() => {
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
// One ordering rule for group conversations, shared by the timeline and the
|
||||
// run trace panel.
|
||||
//
|
||||
// The backend assigns every Agent turn a per-topic `turnSeq` when it is
|
||||
// dispatched (slot order for concurrent replies, dispatch order for
|
||||
// discussions), and stamps failure and system notices with the turn they
|
||||
// belong to. Live bubbles carry the same number as the reply that later
|
||||
// replaces them, so nothing moves when a reply is saved. Older history has
|
||||
// no `turnSeq` and keeps its saved chronological order.
|
||||
|
||||
export function positiveSequence(value) {
|
||||
const number = Number(value)
|
||||
return Number.isSafeInteger(number) && number > 0 ? number : null
|
||||
}
|
||||
|
||||
export function timestampValue(value) {
|
||||
if (typeof value === 'number') return Number.isFinite(value) && value >= 0 ? value : 0
|
||||
const parsed = Date.parse(String(value || ''))
|
||||
return Number.isFinite(parsed) ? parsed : 0
|
||||
}
|
||||
|
||||
export function messageTurnSequence(message) {
|
||||
return positiveSequence(message?.turnSeq)
|
||||
?? positiveSequence(message?.liveAgentRun?.turnSeq)
|
||||
}
|
||||
|
||||
// Returns a sortable key for one timeline message. `topicIndex` places the
|
||||
// message's topic; `index` is the stable position in the source array.
|
||||
export function timelineSortKey(message, { topicIndex, index }) {
|
||||
if (message?.role === 'user' && !message.threadRootId) return [topicIndex, 0, 0, 0, index]
|
||||
const turnSeq = messageTurnSequence(message)
|
||||
if (turnSeq != null) return [topicIndex, 2, turnSeq, 0, index]
|
||||
// Legacy history (and a live bubble whose number has not arrived yet) keeps
|
||||
// chronological order. Live bubbles without a number go after numbered
|
||||
// turns of their topic because they are the newest activity.
|
||||
if (message?.provisional) {
|
||||
return [topicIndex, 3, timestampValue(message.createdAt), 0, index]
|
||||
}
|
||||
return [topicIndex, 1, timestampValue(message?.createdAt), 0, index]
|
||||
}
|
||||
|
||||
export function compareSortKeys(left, right) {
|
||||
const length = Math.max(left.length, right.length)
|
||||
for (let index = 0; index < length; index += 1) {
|
||||
const difference = (left[index] ?? 0) - (right[index] ?? 0)
|
||||
if (difference) return difference
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// Trace panel items: rounds stay grouped, then turn order, then the Agent's
|
||||
// position in the run's own target list (never the mutable group list).
|
||||
export function compareTraceItems(left, right, targetOrder = new Map()) {
|
||||
return (Number(left?.round) || 0) - (Number(right?.round) || 0)
|
||||
|| (positiveSequence(left?.turnSeq) ?? Number.MAX_SAFE_INTEGER)
|
||||
- (positiveSequence(right?.turnSeq) ?? Number.MAX_SAFE_INTEGER)
|
||||
|| (targetOrder.get(left?.agentKind) ?? 99) - (targetOrder.get(right?.agentKind) ?? 99)
|
||||
|| timestampValue(left?.createdAt || left?.startedAt)
|
||||
- timestampValue(right?.createdAt || right?.startedAt)
|
||||
}
|
||||
@@ -413,6 +413,7 @@ export function normalizeRunAgent(value, run = {}) {
|
||||
...(orchestration ? { orchestration } : {}),
|
||||
...(startedAt !== '' ? { startedAt } : {}),
|
||||
...(lastActivityAt !== '' ? { lastActivityAt } : {}),
|
||||
...(Number.isSafeInteger(input.turnSeq) && input.turnSeq > 0 ? { turnSeq: input.turnSeq } : {}),
|
||||
silent: input.silent === true,
|
||||
truncated: input.truncated === true,
|
||||
seenSeqs,
|
||||
|
||||
@@ -289,6 +289,7 @@ function normalizeMessage(value) {
|
||||
const trace = normalizeMessageTrace(input.trace)
|
||||
if (trace) message.trace = trace
|
||||
else delete message.trace
|
||||
if (!Number.isSafeInteger(input.turnSeq) || input.turnSeq <= 0) delete message.turnSeq
|
||||
return message
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user