| ... | ... | @@ -21,6 +21,7 @@ import * as DateTime from "effect/DateTime"; |
| 21 | 21 | import * as Effect from "effect/Effect"; |
| 22 | 22 | import * as Layer from "effect/Layer"; |
| 23 | 23 | import * as Stream from "effect/Stream"; |
| 24 | import * as Tracer from "effect/Tracer"; |
| 24 | 25 | |
| 25 | 26 | import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; |
| 26 | 27 | import * as ServerConfig from "../config.ts"; |
| ... | ... | @@ -749,6 +750,24 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { |
| 749 | 750 | |
| 750 | 751 | // Runtime reconciliation writes restart cancellations under this command |
| 751 | 752 | // prefix, which the live terminal-run listener skips. |
| 753 | // Collects the SQL text of every statement run under the returned tracer. |
| 754 | const traceSqlStatements = () => { |
| 755 | const statements: Array<string> = []; |
| 756 | const tracer = Tracer.make({ |
| 757 | span(options) { |
| 758 | const span = new Tracer.NativeSpan(options); |
| 759 | const end = span.end.bind(span); |
| 760 | span.end = (endTime, exit) => { |
| 761 | end(endTime, exit); |
| 762 | const query = span.attributes.get("db.query.text"); |
| 763 | if (typeof query === "string") statements.push(query); |
| 764 | }; |
| 765 | return span; |
| 766 | }, |
| 767 | }); |
| 768 | return { statements, tracer }; |
| 769 | }; |
| 770 | |
| 752 | 771 | const reconcileCommandId = (name: string) => CommandId.make(`command:runtime-reconcile:${name}`); |
| 753 | 772 | |
| 754 | 773 | const parentProviderThreadId = (threadId: ThreadId) => |
| ... | ... | @@ -1578,6 +1597,111 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { |
| 1578 | 1597 | }), |
| 1579 | 1598 | ); |
| 1580 | 1599 | |
| 1600 | it.effect("settles only the current delivery of a parent with long delivery history", () => |
| 1601 | Effect.gen(function* () { |
| 1602 | const orchestrator = yield* Orchestrator.OrchestratorV2; |
| 1603 | const eventSink = yield* EventSink.EventSinkV2; |
| 1604 | const now = yield* DateTime.now; |
| 1605 | const threadId = ThreadId.make("thread:delivery-history"); |
| 1606 | const runId = RunId.make("run:delivery-history"); |
| 1607 | const taskId = NodeId.make("node:delivery-history-task"); |
| 1608 | const currentMessageId = MessageId.make(`message:delegated-delivery:${threadId}`); |
| 1609 | const currentRunId = RunId.make("run:delivery-history:current"); |
| 1610 | const HISTORY = 40; |
| 1611 | yield* seedParentWithTerminalTask({ |
| 1612 | threadId, |
| 1613 | projectId: ProjectId.make("project:delivery-history"), |
| 1614 | runId, |
| 1615 | rootNodeId: NodeId.make("node:delivery-history-root"), |
| 1616 | taskId, |
| 1617 | deliveryState: "claimed", |
| 1618 | completionWake: "always", |
| 1619 | deliveryTaskIds: [taskId], |
| 1620 | now, |
| 1621 | }); |
| 1622 | const deliveryMessage = (id: MessageId, deliveryRunId: RunId) => ({ |
| 1623 | id: EventId.make(`event:${deliveryRunId}:message`), |
| 1624 | type: "message.updated" as const, |
| 1625 | threadId, |
| 1626 | runId: deliveryRunId, |
| 1627 | occurredAt: now, |
| 1628 | payload: { |
| 1629 | id, |
| 1630 | threadId, |
| 1631 | runId: deliveryRunId, |
| 1632 | nodeId: null, |
| 1633 | role: "user" as const, |
| 1634 | text: `Delegated task ${taskId} reached a terminal state.`, |
| 1635 | attachments: [], |
| 1636 | streaming: false, |
| 1637 | createdBy: "agent" as const, |
| 1638 | creationSource: "server" as const, |
| 1639 | createdAt: now, |
| 1640 | updatedAt: now, |
| 1641 | delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] }, |
| 1642 | }, |
| 1643 | }); |
| 1644 | // Earlier deliveries that already settled. Each one still owns a |
| 1645 | // delegated message and a terminal run. |
| 1646 | const history = Array.from({ length: HISTORY }, (_, index) => { |
| 1647 | const historyRunId = RunId.make(`run:delivery-history:${index}`); |
| 1648 | const messageId = MessageId.make(`message:delivery-history:${index}`); |
| 1649 | return [ |
| 1650 | deliveryMessage(messageId, historyRunId), |
| 1651 | runEvent({ |
| 1652 | threadId, |
| 1653 | runId: historyRunId, |
| 1654 | ordinal: index + 2, |
| 1655 | status: "completed", |
| 1656 | now, |
| 1657 | providerThreadId: parentProviderThreadId(threadId), |
| 1658 | userMessageId: messageId, |
| 1659 | }), |
| 1660 | ]; |
| 1661 | }).flat(); |
| 1662 | yield* eventSink.write({ |
| 1663 | commandId: reconcileCommandId("delivery-history"), |
| 1664 | events: [ |
| 1665 | ...history, |
| 1666 | deliveryMessage(currentMessageId, currentRunId), |
| 1667 | runEvent({ |
| 1668 | threadId, |
| 1669 | runId: currentRunId, |
| 1670 | ordinal: HISTORY + 2, |
| 1671 | status: "completed", |
| 1672 | now, |
| 1673 | providerThreadId: parentProviderThreadId(threadId), |
| 1674 | userMessageId: currentMessageId, |
| 1675 | }), |
| 1676 | ], |
| 1677 | }); |
| 1678 | |
| 1679 | const { statements, tracer } = traceSqlStatements(); |
| 1680 | yield* orchestrator.recoverDelegatedTasks.pipe(Effect.withTracer(tracer)); |
| 1681 | |
| 1682 | const projection = yield* orchestrator.getThreadProjection(threadId); |
| 1683 | assert.equal( |
| 1684 | projection.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, |
| 1685 | "delivered", |
| 1686 | ); |
| 1687 | assert.isNull(projection.runs.find((row) => row.id === runId)?.delegatedCompletion?.delivery); |
| 1688 | // Recovery reads the delivery message once, for the current delivery |
| 1689 | // run only. History must not cost one thread reload per past delivery. |
| 1690 | assert.lengthOf( |
| 1691 | statements.filter((statement) => |
| 1692 | statement.includes("FROM orchestration_v2_projection_messages AS message"), |
| 1693 | ), |
| 1694 | 1, |
| 1695 | ); |
| 1696 | |
| 1697 | // A second recovery finds nothing open on this thread and changes nothing. |
| 1698 | yield* orchestrator.recoverDelegatedTasks; |
| 1699 | const again = yield* orchestrator.getThreadProjection(threadId); |
| 1700 | assert.deepEqual(again.runs, projection.runs); |
| 1701 | assert.deepEqual(again.subagents, projection.subagents); |
| 1702 | }), |
| 1703 | ); |
| 1704 | |
| 1581 | 1705 | it.effect("settles a cancelled child whose held wakes wait behind it", () => |
| 1582 | 1706 | Effect.gen(function* () { |
| 1583 | 1707 | const orchestrator = yield* Orchestrator.OrchestratorV2; |
Theo Browne <me@t3.gg> 2026-10-14 15:53:07-07:00
GitHub <noreply@github.com> 2026-10-14 15:53:07-07:00