Port plan and todos as world extensions
This commit is contained in:
@@ -0,0 +1,19 @@
|
||||
# Plan
|
||||
|
||||
The `plan` tool maintains a durable epic/task ledger in `ultimate.plan`.
|
||||
Task-owned descendants read their top owner's ledger. Only that owner edits it.
|
||||
Ownerless conversations maintain separate ledgers.
|
||||
|
||||
Edits and receipts commit together. Memo namespaces distinguish nested calls that share a task.
|
||||
The tool returns both JSON text and structured content.
|
||||
The compact prompt section excludes completed epics.
|
||||
|
||||
The root document `ultimate.plan.tasks` indexes linked child conversations.
|
||||
A new link or task title change updates the child's metadata title.
|
||||
Other edits preserve a title that the user supplied.
|
||||
|
||||
The first input of a new lineage branch clears its inherited plan atomically.
|
||||
The hook preserves a branch's plan if the branch already answered.
|
||||
The reader uses the existing `lineage.branch` document contract without requiring the lineage extension.
|
||||
|
||||
`host/tests/plan_todos.rs` verifies the port through the world extension loader.
|
||||
@@ -0,0 +1,315 @@
|
||||
/** The durable epic/task ledger, linked-child index, and first-input branch reset. */
|
||||
import { arr, int, obj, oneOf, opt, str } from '@earendil-works/pi-schema'
|
||||
import {
|
||||
defineDoc,
|
||||
defineExtension,
|
||||
defineTool,
|
||||
GenerationTask,
|
||||
hook,
|
||||
ROOT_CONVERSATION_ID,
|
||||
section,
|
||||
} from '@earendil-works/pi-durable'
|
||||
import type { ConversationDocToken, ConversationId, Cursor, EntryId, JsonObject, Tx } from '@earendil-works/pi-durable'
|
||||
// Read lineage's document contract without requiring its extension to exist yet.
|
||||
export const Branch = defineDoc<{ branch?: { from: number; at: number; intent: string; brief?: true } }>({
|
||||
kind: 'lineage.branch', checkpointWhen: () => false, version: 1,
|
||||
scope: 'conversation', history: 'latest', fork: 'initial', initial: () => ({}),
|
||||
})
|
||||
type ExtensionHost = { metadata: ConversationDocToken<{ title: string; label: string }> }
|
||||
|
||||
export type Task = {
|
||||
key: string
|
||||
title: string
|
||||
role: 'worker' | 'reviewer'
|
||||
description: string
|
||||
deps: string[]
|
||||
status: 'backlog' | 'todo' | 'in_progress' | 'in_review' | 'done' | 'blocked' | 'canceled'
|
||||
child?: number
|
||||
verdict?: string
|
||||
notes: { at: string; text: string }[]
|
||||
}
|
||||
export type Epic = {
|
||||
key: string
|
||||
title: string
|
||||
body: string
|
||||
status: 'planning' | 'in_progress' | 'done' | 'blocked'
|
||||
verdict?: string
|
||||
tasks: Record<string, Task>
|
||||
}
|
||||
export const Plan = defineDoc<{ next: number; epics: Record<string, Epic> }>({
|
||||
kind: 'ultimate.plan',
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'rewindable',
|
||||
fork: 'asOf',
|
||||
initial: () => ({ next: 1, epics: {} }),
|
||||
})
|
||||
/** The task each linked child conversation works on, across every coordinator's plan; kept with the root. */
|
||||
export type TaskLink = {
|
||||
coordinator: number
|
||||
epic: string
|
||||
key: string
|
||||
title: string
|
||||
status: Task['status']
|
||||
}
|
||||
export const TaskIndex = defineDoc<{ children: Record<string, TaskLink> }>({
|
||||
kind: 'ultimate.plan.tasks',
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
initial: () => ({ children: {} }),
|
||||
})
|
||||
export const docs = [Plan, TaskIndex]
|
||||
// Commit the receipt with the edit, closing the crash window before api.memo stores the result.
|
||||
const Receipt = defineDoc<{ results: Record<string, string> }>({
|
||||
kind: 'ultimate.plan.receipt',
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'task',
|
||||
initial: () => ({ results: {} }),
|
||||
})
|
||||
// structuredContent hands codemode scripts the object itself. As text, `shown.description` was undefined, and
|
||||
// `update {description: shown.description + more}` wrote "undefined" over the description (#159728).
|
||||
const reply = (text: string) => ({
|
||||
output: [{ type: 'text' as const, text }],
|
||||
structuredContent: JSON.parse(text) as JsonObject,
|
||||
})
|
||||
const required = (value: string | undefined, field: string) => {
|
||||
if (value === undefined) throw new Error(`${field} is required`)
|
||||
return value
|
||||
}
|
||||
|
||||
/**
|
||||
* Rewrite the index entries of `coordinator` from its plan. A link that is new or whose task title changed renames
|
||||
* the child after its task.
|
||||
*/
|
||||
async function reindex(
|
||||
tx: Tx,
|
||||
host: Pick<ExtensionHost, 'metadata'>,
|
||||
coordinator: ConversationId,
|
||||
plan: { epics: Record<string, Epic> },
|
||||
) {
|
||||
const index = await tx.doc(TaskIndex, ROOT_CONVERSATION_ID)
|
||||
const previous = { ...index.children }
|
||||
for (const [child, link] of Object.entries(previous))
|
||||
if (link.coordinator === coordinator) delete index.children[child]
|
||||
for (const epic of Object.values(plan.epics)) {
|
||||
for (const task of Object.values(epic.tasks)) {
|
||||
if (task.child === undefined) continue
|
||||
const child = String(task.child)
|
||||
const before = previous[child]
|
||||
index.children[child] = {
|
||||
coordinator,
|
||||
epic: epic.key,
|
||||
key: task.key,
|
||||
title: task.title,
|
||||
status: task.status,
|
||||
}
|
||||
const renamed =
|
||||
before?.coordinator !== coordinator ||
|
||||
before.key !== task.key ||
|
||||
before.title !== task.title
|
||||
if (renamed && (await tx.conversation(task.child as ConversationId)))
|
||||
(await tx.doc(host.metadata, task.child as ConversationId)).title = task.title
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Set by a conversation's first run; a fork starts without it. */
|
||||
export const Started = defineDoc<{ started?: true }>({
|
||||
kind: 'plan.started',
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
initial: () => ({}),
|
||||
})
|
||||
|
||||
/** Whether branch `id` has answered since it forked at entry `at`. */
|
||||
async function answered(tx: Tx, id: ConversationId, at: number) {
|
||||
let cursor: Cursor | undefined
|
||||
do {
|
||||
const page = await tx.scanEntries(
|
||||
{ conversationId: id, minEntryId: (at + 1) as EntryId, order: 'ascending' },
|
||||
64,
|
||||
cursor,
|
||||
)
|
||||
if (page.items.some((entry) => entry.kind === 'pi.assistant')) return true
|
||||
cursor = page.next
|
||||
} while (cursor !== undefined)
|
||||
return false
|
||||
}
|
||||
|
||||
export default (host: ExtensionHost) => {
|
||||
return defineExtension({
|
||||
name: 'plan',
|
||||
hooks: [
|
||||
// Pi upstream: @earendil-works/pi-durable@1.5.0 — patched transactional beforeInput hook.
|
||||
// A lineage branch starts with an empty plan, not the one it forked with: emptied in the commit that places its
|
||||
// first input, so its first request never sees the inherited plan. A branch that answered before it was marked
|
||||
// started (one made before this hook) keeps its own.
|
||||
hook(GenerationTask, {
|
||||
beforeInput: async (_input, { tx, conversationId }) => {
|
||||
if ((await tx.doc(Started, conversationId)).started) return undefined
|
||||
const branch = (await tx.doc(Branch, conversationId)).branch
|
||||
const reset = !!branch && !(await answered(tx, conversationId, branch.at))
|
||||
;(await tx.doc(Started, conversationId)).started = true
|
||||
if (reset) Object.assign(await tx.doc(Plan, conversationId), Plan.definition.initial())
|
||||
return undefined
|
||||
},
|
||||
}),
|
||||
],
|
||||
sections: [
|
||||
section('plan', async ({ read, conversationId }, context) => {
|
||||
const plan = await read.snapshot(Plan, conversationId, context)
|
||||
const lines = Object.values(plan?.epics ?? {})
|
||||
.filter((epic) => epic.status !== 'done')
|
||||
.flatMap((epic) => [
|
||||
`${epic.key} ${epic.status} ${epic.title}`,
|
||||
...Object.values(epic.tasks).map(
|
||||
(task) =>
|
||||
`${task.key} ${task.status} ${task.title}` +
|
||||
(task.deps.length ? ` deps:${task.deps.join(',')}` : '') +
|
||||
(task.child !== undefined ? ` child:${task.child}` : ''),
|
||||
),
|
||||
])
|
||||
return lines.length ? lines.join('\n') : undefined
|
||||
}),
|
||||
],
|
||||
tools: [
|
||||
defineTool({
|
||||
name: 'plan',
|
||||
description:
|
||||
'Maintain the durable epic/task ledger. The tool assigns keys: epics E1, E2; tasks E1.1, E1.2. Use only keys a call has returned: wait for create/add results before calls that reference them (epic, deps, key), never in the same parallel batch. create {title,body}; add {epic,title,role,description,deps?}; update {key,status?,title?,description?,body?,child?,verdict?} replaces only the fields given (description and body whole: show first to extend them); in codemode every action returns the object; note {key,text} notes a task only (E1.1, not E1; record on an epic with update verdict or body); show {key?} reads an epic, task, or the whole ledger. Epic bodies use Request, Selected approach, Intent, Scope (in/out), binary Acceptance criteria, Implementation, Verification, Risks. Task descriptions use Outcome, Scope, Non-goals, Files, Acceptance criteria with a verification command, Constraints. Epic statuses: planning, in_progress, done, blocked. Task statuses: backlog, todo, in_progress, in_review, done, blocked, canceled. child links a conversation number; deps are task keys. body is epic-only; verdict records evidence on either an epic or a task. Children read the root coordinator’s ledger with show; only that coordinator can edit it. Read full plans with show; the prompt contains only a compact ledger.',
|
||||
// A memoized receipt id records the result in the same transaction as the ledger change.
|
||||
replay: 'safe',
|
||||
parameters: obj({
|
||||
action: oneOf(['create', 'add', 'update', 'note', 'show'] as const),
|
||||
title: opt(str()),
|
||||
body: opt(str()),
|
||||
epic: opt(str({ description: 'Epic key returned by create, e.g. E1.' })),
|
||||
role: opt(oneOf(['worker', 'reviewer'] as const)),
|
||||
description: opt(str()),
|
||||
deps: opt(
|
||||
arr(str(), {
|
||||
description: 'Keys of tasks already added, e.g. ["E1.1"].',
|
||||
}),
|
||||
),
|
||||
key: opt(
|
||||
str({
|
||||
description: 'A returned key: epic E1 or task E1.1; note takes a task key.',
|
||||
}),
|
||||
),
|
||||
status: opt(str()),
|
||||
child: opt(int({ minimum: 1 })),
|
||||
verdict: opt(str()),
|
||||
text: opt(str()),
|
||||
}),
|
||||
execute: async (args, api, context) => {
|
||||
const cached = await api.memo<ReturnType<typeof reply>>('result', context)
|
||||
if (cached) return cached
|
||||
const receiptId = await api.memo('receipt-id', crypto.randomUUID(), context)
|
||||
const text = await api.commit(async (tx) => {
|
||||
let coordinator = api.conversationId
|
||||
let owner = (await tx.conversation(coordinator))?.owner
|
||||
while (owner) {
|
||||
coordinator = owner.conversationId
|
||||
owner = (await tx.conversation(coordinator))?.owner
|
||||
}
|
||||
if (args.action !== 'show' && coordinator !== api.conversationId)
|
||||
throw new Error(
|
||||
`Only conversation ${coordinator} edits this plan; report to it instead`,
|
||||
)
|
||||
const receipt = await tx.doc(Receipt, api.taskId)
|
||||
if (receipt.results[receiptId] !== undefined) return receipt.results[receiptId]
|
||||
const plan = await tx.doc(Plan, coordinator)
|
||||
const lookup = (key: string) => {
|
||||
const epicKey = key.split('.')[0]
|
||||
const epic = Object.hasOwn(plan.epics, epicKey) ? plan.epics[epicKey] : undefined
|
||||
const item = key.includes('.')
|
||||
? epic && Object.hasOwn(epic.tasks, key)
|
||||
? epic.tasks[key]
|
||||
: undefined
|
||||
: epic
|
||||
if (!item) throw new Error(`Unknown key: ${key}`)
|
||||
return item
|
||||
}
|
||||
let result: unknown
|
||||
if (args.action === 'create') {
|
||||
const title = required(args.title, 'title')
|
||||
const body = required(args.body, 'body')
|
||||
const key = `E${plan.next++}`
|
||||
plan.epics[key] = {
|
||||
key,
|
||||
title,
|
||||
body,
|
||||
status: 'planning',
|
||||
tasks: {},
|
||||
}
|
||||
result = plan.epics[key]
|
||||
} else if (args.action === 'add') {
|
||||
const epic = lookup(required(args.epic, 'epic'))
|
||||
if (!('tasks' in epic)) throw new Error('epic must identify an epic')
|
||||
const title = required(args.title, 'title')
|
||||
const description = required(args.description, 'description')
|
||||
if (!args.role) throw new Error('role is required')
|
||||
for (const dep of args.deps ?? []) {
|
||||
if (!('notes' in lookup(dep))) throw new Error('deps must identify tasks')
|
||||
}
|
||||
const key = `${epic.key}.${Object.keys(epic.tasks).length + 1}`
|
||||
epic.tasks[key] = {
|
||||
key,
|
||||
title,
|
||||
description,
|
||||
role: args.role,
|
||||
deps: args.deps ?? [],
|
||||
status: 'backlog',
|
||||
notes: [],
|
||||
}
|
||||
result = epic.tasks[key]
|
||||
} else if (args.action === 'show') {
|
||||
result = args.key === undefined ? plan : lookup(args.key)
|
||||
} else {
|
||||
const item = lookup(required(args.key, 'key'))
|
||||
const task = 'notes' in item
|
||||
if (args.action === 'note') {
|
||||
if (!task) throw new Error('note requires a task key')
|
||||
item.notes.push({
|
||||
at: new Date().toISOString(),
|
||||
text: required(args.text, 'text'),
|
||||
})
|
||||
} else {
|
||||
const statuses = task
|
||||
? ['backlog', 'todo', 'in_progress', 'in_review', 'done', 'blocked', 'canceled']
|
||||
: ['planning', 'in_progress', 'done', 'blocked']
|
||||
if (args.status !== undefined && !statuses.includes(args.status))
|
||||
throw new Error(`Invalid status for ${item.key}: ${args.status}`)
|
||||
if (task && args.body !== undefined) throw new Error('body requires an epic key')
|
||||
if (!task && (args.description !== undefined || args.child !== undefined))
|
||||
throw new Error('description and child require a task key')
|
||||
if (args.status !== undefined) item.status = args.status as typeof item.status
|
||||
if (args.title !== undefined) item.title = args.title
|
||||
if (args.verdict !== undefined) item.verdict = args.verdict
|
||||
if (task) {
|
||||
if (args.description !== undefined) item.description = args.description
|
||||
if (args.child !== undefined) item.child = args.child
|
||||
} else {
|
||||
if (args.body !== undefined) item.body = args.body
|
||||
}
|
||||
}
|
||||
result = item
|
||||
}
|
||||
if (args.action !== 'show') await reindex(tx, host, coordinator, plan)
|
||||
receipt.results[receiptId] = JSON.stringify(result)
|
||||
return receipt.results[receiptId]
|
||||
}, context)
|
||||
return api.memo('result', reply(text), context)
|
||||
},
|
||||
}),
|
||||
].map((tool) => ({ ...tool, outputSchema: { type: 'object' } })),
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
# Todos
|
||||
|
||||
The root document `todos` stores the world's to-do list.
|
||||
Each item belongs to `user` or `ultimate`.
|
||||
The `todo` tool adds items, completes items, and lists open items by owner.
|
||||
Text must contain 1–300 characters after the tool trims it.
|
||||
The list retains the newest 50 completed items. The exported `mark` helper also reopens items.
|
||||
|
||||
Only the standing `ultimate` conversation uses the tool.
|
||||
The reader uses the existing `ultimate.standing-agents` document contract.
|
||||
Until subagents supplies that document, no conversation qualifies automatically.
|
||||
This port does not designate the root as `ultimate`.
|
||||
|
||||
The `beforeInput` hook places a note only when the standing conversation's open list changes.
|
||||
The note and the remembered list commit with the input.
|
||||
The tool remains deferred. A caller must load it through `modelTools` or a tool-search extension.
|
||||
|
||||
This port contains no browser actions or side-chat delegation.
|
||||
`host/tests/plan_todos.rs` verifies the four original cases and real-loader tool calls.
|
||||
@@ -0,0 +1,176 @@
|
||||
/** The world's root to-do list, its todo tool, and changed-list notes for the standing @ultimate conversation. */
|
||||
import { int, obj, oneOf, opt, str } from '@earendil-works/pi-schema'
|
||||
import {
|
||||
defineDoc,
|
||||
defineExtension,
|
||||
defineTool,
|
||||
GenerationTask,
|
||||
hook,
|
||||
ROOT_CONVERSATION_ID,
|
||||
} from '@earendil-works/pi-durable'
|
||||
import type { ConversationId, Tx } from '@earendil-works/pi-durable'
|
||||
// Read subagents' standing-agent contract without requiring its extension yet.
|
||||
export const Standing = defineDoc<Record<string, ConversationId>>({
|
||||
kind: 'ultimate.standing-agents', checkpointWhen: () => false, version: 1,
|
||||
scope: 'conversation', history: 'latest', fork: 'initial', initial: () => ({}),
|
||||
})
|
||||
export type Todo = { id: number; text: string; owner: 'user' | 'ultimate'; at: number; done?: number }
|
||||
|
||||
/** Done to-dos kept; older ones are dropped when another is ticked. */
|
||||
export const DONE_LIMIT = 50
|
||||
const TEXT_CHARS = 300
|
||||
|
||||
export const Todos = defineDoc<{ next: number; items: Todo[] }>({
|
||||
kind: 'todos',
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
initial: () => ({ next: 1, items: [] }),
|
||||
})
|
||||
/** The open list @ultimate last heard, so it hears again only after a change. */
|
||||
const Seen = defineDoc<{ key?: string }>({
|
||||
kind: 'todos.seen',
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
initial: () => ({}),
|
||||
})
|
||||
export const docs = [Todos]
|
||||
|
||||
const OWNERS = ['user', 'ultimate'] as const
|
||||
|
||||
export async function add(tx: Tx, text: unknown, owner: unknown) {
|
||||
if (typeof text !== 'string' || !text.trim()) throw new Error('text must be nonempty')
|
||||
if (text.trim().length > TEXT_CHARS)
|
||||
throw new Error(`text must be at most ${TEXT_CHARS} characters`)
|
||||
if (!OWNERS.includes(owner as Todo['owner']))
|
||||
throw new Error("owner must be 'user' or 'ultimate'")
|
||||
const todos = await tx.doc(Todos, ROOT_CONVERSATION_ID)
|
||||
const todo: Todo = {
|
||||
id: todos.next++,
|
||||
text: text.trim(),
|
||||
owner: owner as Todo['owner'],
|
||||
at: Date.now(),
|
||||
}
|
||||
todos.items.push(todo)
|
||||
return todo
|
||||
}
|
||||
|
||||
/** Tick to-do `id` (or open it again), keeping only the newest `DONE_LIMIT` done ones. */
|
||||
export async function mark(tx: Tx, id: unknown, done: boolean) {
|
||||
const todos = await tx.doc(Todos, ROOT_CONVERSATION_ID)
|
||||
const todo = todos.items.find((item) => item.id === id)
|
||||
if (!todo) throw new Error(`No to-do ${String(id)}`)
|
||||
if (done) {
|
||||
todo.done ??= Date.now()
|
||||
const ticked = todos.items.filter((item) => item.done !== undefined)
|
||||
const drop = new Set(
|
||||
ticked
|
||||
.toSorted((a, b) => b.done! - a.done!)
|
||||
.slice(DONE_LIMIT)
|
||||
.map((item) => item.id),
|
||||
)
|
||||
todos.items = todos.items.filter((item) => !drop.has(item.id))
|
||||
} else delete todo.done
|
||||
// A copy: the document's overlay is unusable once the commit settles, and callers reply with this.
|
||||
return { ...todo }
|
||||
}
|
||||
|
||||
const line = (todo: Todo) => `- ${todo.id}: ${todo.text}`
|
||||
/** The open to-dos as model text, by owner; undefined when none is open. */
|
||||
export function listText(items: readonly Todo[]) {
|
||||
const open = items.filter((item) => item.done === undefined)
|
||||
if (!open.length) return undefined
|
||||
const mine = open.filter((item) => item.owner === 'ultimate')
|
||||
const theirs = open.filter((item) => item.owner === 'user')
|
||||
return [
|
||||
...(mine.length ? ['Yours:', ...mine.map(line)] : []),
|
||||
...(theirs.length ? ["The user's:", ...theirs.map(line)] : []),
|
||||
].join('\n')
|
||||
}
|
||||
|
||||
/** @ultimate's note when the open list changed since it last heard it; undefined otherwise. */
|
||||
export async function noteFor(tx: Tx, conversationId: ConversationId, now: number) {
|
||||
const items = (await tx.doc(Todos, ROOT_CONVERSATION_ID)).items
|
||||
const text = listText(items)
|
||||
const key = text ?? ''
|
||||
const seen = await tx.doc(Seen, conversationId)
|
||||
if ((seen.key ?? '') === key) return undefined
|
||||
seen.key = key
|
||||
return {
|
||||
entries: [
|
||||
{
|
||||
kind: 'todos.note',
|
||||
data: { open: items.filter((item) => item.done === undefined).length },
|
||||
model: [
|
||||
{
|
||||
role: 'user' as const,
|
||||
content: text
|
||||
? `Open to-dos (todo tool: done when finished):\n${text}`
|
||||
: 'No to-dos are open.',
|
||||
timestamp: now,
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
const standing = (tx: Tx, id: ConversationId) =>
|
||||
tx.doc(Standing, ROOT_CONVERSATION_ID).then((doc) => doc.ultimate === id)
|
||||
|
||||
export default function todosExtension() {
|
||||
return defineExtension({
|
||||
name: 'todos',
|
||||
hooks: [
|
||||
hook(GenerationTask, {
|
||||
// Pi upstream: @earendil-works/pi-durable@1.5.0 — patched transactional beforeInput hook.
|
||||
// A per-input note when the list changed, never a system-prompt section, keeps the prompt cached.
|
||||
beforeInput: async (_input, { tx, conversationId }) =>
|
||||
(await standing(tx, conversationId))
|
||||
? noteFor(tx, conversationId, Date.now())
|
||||
: undefined,
|
||||
}),
|
||||
],
|
||||
tools: [
|
||||
defineTool({
|
||||
name: 'todo',
|
||||
deferred: true,
|
||||
description:
|
||||
"The world's to-dos, owned by you ('ultimate') or the user ('user'): add {text, owner}, done {id}, list. Keep open loops here. Only the standing @ultimate conversation maintains this list.",
|
||||
parameters: obj({
|
||||
action: oneOf(['add', 'done', 'list']),
|
||||
text: opt(str()),
|
||||
owner: opt(oneOf(['user', 'ultimate'])),
|
||||
id: opt(int({ minimum: 1 })),
|
||||
}),
|
||||
// Not replay-safe (the default): an interrupted add is never rerun, so it commits at most once.
|
||||
async execute(args, api, ctx) {
|
||||
// Side chats are not part of this port. Each caller acts as itself.
|
||||
const caller = api.conversationId
|
||||
const text = await api.commit(async (tx) => {
|
||||
// Deferred only hides it: a branch can still find it with tool_search, so the caller is checked here.
|
||||
if (!(await standing(tx, caller)))
|
||||
throw new Error(
|
||||
'Only @ultimate keeps the to-dos; ask @ultimate to add one.',
|
||||
)
|
||||
if (args.action === 'add') {
|
||||
const todo = await add(tx, args.text, args.owner)
|
||||
return `Added ${todo.id} for ${todo.owner}: ${todo.text}`
|
||||
}
|
||||
if (args.action === 'done') {
|
||||
const todo = await mark(tx, args.id, true)
|
||||
return `Done ${todo.id}: ${todo.text}`
|
||||
}
|
||||
return listText((await tx.doc(Todos, ROOT_CONVERSATION_ID)).items) ?? 'No open to-dos.'
|
||||
}, ctx)
|
||||
return { output: [{ type: 'text', text }] }
|
||||
},
|
||||
}),
|
||||
],
|
||||
})
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js" },
|
||||
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js && esbuild plan-todos-test.ts --bundle --format=esm --platform=browser --outfile=plan-todos-test-bundle.js" },
|
||||
"dependencies": {
|
||||
"@earendil-works/pi-durable": "1.5.0",
|
||||
"@earendil-works/pi-ai": "1.5.0",
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,151 @@
|
||||
/** The six Ultimate plan/todos cases, plus real-loader turns and branch reset. */
|
||||
import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
|
||||
import { createModels } from '@earendil-works/pi-ai/models';
|
||||
import { fauxProvider, fauxAssistantMessage, fauxToolCall } from '@earendil-works/pi-ai/providers/faux';
|
||||
import { Harness, createRegistry, defineExtension } from '@earendil-works/pi-durable';
|
||||
import { openStorage } from './sqlite';
|
||||
import { loadExtensions, fs, Metadata } from './extensions';
|
||||
const assert = (value:any, message:string) => { if (!value) throw new Error(message); };
|
||||
const equal = (actual:any, expected:any) => assert(JSON.stringify(actual) === JSON.stringify(expected), `Mismatch: ${JSON.stringify(actual)} != ${JSON.stringify(expected)}`);
|
||||
const rejects = async (call:()=>Promise<any>, text:string) => { try { await call(); } catch(error) { assert(String(error).includes(text), String(error)); return; } throw new Error(`Expected ${text}`); };
|
||||
const models = createModels(), faux = fauxProvider({tokensPerSecond:100000}); models.setProvider(faux.provider);
|
||||
const registry = createRegistry(), storage = await openStorage(globalThis.storagePath);
|
||||
const harness = await Harness.open(storage,{models,registry,settings:{get extensions(){return registry.snapshot().installed();}}},context);
|
||||
const root = await harness.root(context,{agent:{model:{provider:'faux',modelId:'faux-1'}}});
|
||||
const loader = await loadExtensions(harness,registry,storage,models,context);
|
||||
assert(!loader.state().error, loader.state().error);
|
||||
const base = `file:///run/sausage/extensions/${loader.state().good}/extensions`;
|
||||
const planModule = await import(/* @vite-ignore */ `${base}/plan/server.ts`);
|
||||
const todosModule = await import(/* @vite-ignore */ `${base}/todos/server.ts`);
|
||||
const {Plan,TaskIndex,Branch,Started} = planModule;
|
||||
const {Todos,Standing,add,mark,noteFor,DONE_LIMIT} = todosModule;
|
||||
const call = async (args:any, id=root.id, name='plan') => {
|
||||
const conversation = await harness.conversation(id,context);
|
||||
await conversation.configure({model:{provider:'faux',modelId:'faux-1'},modelTools:{add:[todosModule.default().tools[0]]}},context);
|
||||
faux.setResponses([fauxAssistantMessage([fauxToolCall(name,args)],{stopReason:'toolUse'}),fauxAssistantMessage('Complete')]);
|
||||
await (await conversation.submit({type:'input',content:'Update ledger'},context)).wait(context);
|
||||
await harness.waitForIdle(context);
|
||||
return (await conversation.context(context)).entries.filter((entry:any)=>entry.kind==='pi.tool-result').at(-1).model[0];
|
||||
};
|
||||
const text = (result:any)=>result.content.map((block:any)=>block.text).join('');
|
||||
harness.resume();
|
||||
try {
|
||||
if (globalThis.planTodosRestart) {
|
||||
const saved = JSON.parse(await fs.read('/saved'));
|
||||
equal(await harness.snapshot(Plan,root.id,context),saved.plan);
|
||||
equal(await harness.snapshot(TaskIndex,root.id,context),saved.index);
|
||||
equal(await harness.snapshot(Todos,root.id,context),saved.todos);
|
||||
assert(!(await call({action:'show'})).isError,'Plan survives restart');
|
||||
} else {
|
||||
// Plan case 1: descendants, verdicts, memo replay, transactional receipts, nested namespaces.
|
||||
const extension = planModule.default({metadata:Metadata});
|
||||
const tool = extension.tools[0], replayResults:any[] = [], nestedResults:any[] = [];
|
||||
let childId:number, grandchildId:number;
|
||||
registry.install(defineExtension({...extension,tools:[{...tool,execute:async(args:any,api:any,ctx:any)=>{
|
||||
if(args.title==='Nested') {
|
||||
const results=[];
|
||||
for(const namespace of ['first','second']) {
|
||||
const nestedApi={...api,memo:(name:string,value:any,memoContext?:any)=>memoContext===undefined?api.memo(`${namespace}/${name}`,ctx):api.memo(`${namespace}/${name}`,value,memoContext)};
|
||||
const first=await tool.execute(args,nestedApi,ctx);
|
||||
const replayApi={...nestedApi,memo:(name:string,...rest:any[])=>name==='result'&&rest.length===1?Promise.resolve(undefined):nestedApi.memo(name,...rest)};
|
||||
nestedResults.push([first,await tool.execute(args,replayApi,ctx)]);results.push(first);
|
||||
}
|
||||
return results[1];
|
||||
}
|
||||
const result=await tool.execute(args,api,ctx);
|
||||
if(api.conversationId===childId && grandchildId===undefined) grandchildId=await api.commit(async(tx:any)=>(await tx.createConversation({ownership:{kind:'task',taskId:api.taskId}})).id,ctx);
|
||||
if(args.action==='create') childId=await api.commit(async(tx:any)=>(await tx.createConversation({ownership:{kind:'task',taskId:api.taskId}})).id,ctx);
|
||||
if(args.action==='add') {
|
||||
const replayApi={...api,memo:(name:string,...rest:any[])=>name==='result'&&rest.length===1?Promise.resolve(undefined):api.memo(name,...rest)};
|
||||
replayResults.push([result,await tool.execute(args,api,ctx),await tool.execute(args,replayApi,ctx)]);
|
||||
}
|
||||
return result;
|
||||
}}]}));
|
||||
await call({action:'create',title:'Ship ledger',body:'## Request\nDurable plans'});
|
||||
await call({action:'add',epic:'E1',title:'Implement',role:'worker',description:'Build it'});
|
||||
await call({action:'add',epic:'E1',title:'Review',role:'reviewer',description:'Check it',deps:['E1.1']});
|
||||
await call({action:'update',key:'E1.1',status:'in_review',child:42,verdict:'Checks passed'});
|
||||
await call({action:'note',key:'E1.1',text:'Checks passed'});
|
||||
const saved=await harness.snapshot(Plan,root.id,context), task=saved.epics.E1.tasks['E1.1'];
|
||||
equal(Object.keys(saved.epics.E1.tasks),['E1.1','E1.2']);
|
||||
assert(saved.next===2 && task.status==='in_review' && task.child===42 && task.verdict==='Checks passed','Saved verdict');
|
||||
assert(task.notes[0].text==='Checks passed' && typeof task.notes[0].at==='string','Saved note');
|
||||
equal(saved.epics.E1.tasks['E1.2'].deps,['E1.1']);
|
||||
assert(saved.epics.E1.tasks['E1.2'].role==='reviewer','Reviewer role');
|
||||
assert(replayResults.length===2,'Two replayed additions');
|
||||
for(const [result,memo,receipt] of replayResults){equal(memo,result);equal(receipt,result);}
|
||||
for(const key of ['E1','E1.1']) equal(JSON.parse(text(await call({action:'show',key},childId!))),key==='E1'?saved.epics.E1:task);
|
||||
for(const args of [{action:'create',title:'Forbidden',body:'No'},{action:'add',epic:'E1',title:'Forbidden',role:'worker',description:'No'},{action:'update',key:'E1.1',status:'done'},{action:'note',key:'E1.1',text:'No'}]) {
|
||||
const denied=await call(args,childId!);assert(denied.isError && text(denied).includes(`Only conversation ${root.id} edits this plan`),'Descendant write refusal');
|
||||
}
|
||||
equal(JSON.parse(text(await call({action:'show',key:'E1.1'},grandchildId!))),task);
|
||||
assert((await call({action:'note',key:'E1.1',text:'No'},grandchildId!)).isError,'Grandchild write refusal');
|
||||
equal(await harness.snapshot(Plan,root.id,context),saved);
|
||||
for(const args of [{action:'update',key:'E1.1',body:'No'},{action:'update',key:'E1',status:'todo'},{action:'show',key:'E99'}]) assert((await call(args)).isError,'Invalid plan edit');
|
||||
equal(await harness.snapshot(Plan,root.id,context),saved);
|
||||
await call({action:'update',key:'E1',status:'done',verdict:'Approved'});await call({action:'show'});
|
||||
await call({action:'add',epic:'E1',title:'Nested',role:'worker',description:'Separate calls'});
|
||||
assert(nestedResults.length===2 && JSON.stringify(nestedResults[0][0])!==JSON.stringify(nestedResults[1][0]),'Separate nested calls');
|
||||
for(const [first,replay] of nestedResults)equal(first,replay);
|
||||
equal(Object.keys((await harness.snapshot(Plan,root.id,context)).epics.E1.tasks),['E1.1','E1.2','E1.3','E1.4']);
|
||||
registry.install(extension);
|
||||
|
||||
// Plan case 2: the child index tracks title changes and link replacement.
|
||||
await harness.commit(async(tx:any)=>Object.assign(await tx.doc(Plan,root.id),Plan.definition.initial()),context);
|
||||
const [first,second]=await harness.commit(async(tx:any)=>[(await tx.createConversation({ownership:{kind:'ownerless'}})).id,(await tx.createConversation({ownership:{kind:'ownerless'}})).id],context);
|
||||
await call({action:'create',title:'Index',body:'Body'});
|
||||
await call({action:'add',epic:'E1',title:'Build it',role:'worker',description:'D'});
|
||||
await call({action:'update',key:'E1.1',child:first});
|
||||
const index=async()=>(await harness.snapshot(TaskIndex,root.id,context)).children;
|
||||
const title=async(id:number)=>(await harness.snapshot(Metadata,id,context)).title;
|
||||
equal(await index(),{[first]:{coordinator:root.id,epic:'E1',key:'E1.1',title:'Build it',status:'backlog'}});equal(await title(first),'Build it');
|
||||
await call({action:'update',key:'E1.1',status:'in_progress',title:'Build it well'});
|
||||
assert((await index())[first].status==='in_progress','Index status');equal(await title(first),'Build it well');
|
||||
await harness.commit(async(tx:any)=>{(await tx.doc(Metadata,first)).title='Renamed by hand';},context);
|
||||
await call({action:'note',key:'E1.1',text:'Progress'});equal(await title(first),'Renamed by hand');
|
||||
await call({action:'update',key:'E1.1',child:second});equal(Object.keys(await index()),[String(second)]);equal(await title(second),'Build it well');
|
||||
|
||||
// A new lineage branch clears the inherited plan in its first input commit.
|
||||
const at=await root.commit((tx:any)=>tx.appendEntry(root.id,{kind:'branch.origin',data:{}}),context);
|
||||
const branch=await root.commit(async(tx:any)=>{const record=await tx.forkConversation(root.id,at.id,{ownership:{kind:'ownerless'}});(await tx.doc(Branch,record.id)).branch={from:root.id,at:at.id,intent:'Fresh'};return record;},context);
|
||||
assert(Object.keys((await harness.snapshot(Plan,branch.id,context)).epics).length===1,'Fork inherited plan');
|
||||
equal(JSON.parse(text(await call({action:'show'},branch.id))),Plan.definition.initial());
|
||||
assert((await harness.snapshot(Started,branch.id,context)).started,'Branch started');
|
||||
|
||||
// Todos cases 1-4 retain the original small transaction fixture.
|
||||
const ULTIMATE=5;
|
||||
const fakeTx=()=>{const docs=new Map();return {doc:async({definition:def}:any,id:number)=>{const key=`${def.kind}:${id}`;if(!docs.has(key))docs.set(key,def.kind==='ultimate.standing-agents'?{ultimate:ULTIMATE}:def.initial());return docs.get(key);}};};
|
||||
const list=async(tx:any)=>(await tx.doc(Todos,1)).items;
|
||||
{
|
||||
const tx=fakeTx();equal((await add(tx,' Deploy fleet ','ultimate')).text,'Deploy fleet');
|
||||
assert((await add(tx,'Rotate the key','user')).id===2,'Sequential todo ids');
|
||||
await rejects(()=>add(tx,' ','user'),'nonempty');await rejects(()=>add(tx,'x','lander'),'owner');equal((await list(tx)).map((todo:any)=>todo.id),[1,2]);
|
||||
}
|
||||
{
|
||||
const tx=fakeTx();for(let i=0;i<DONE_LIMIT+2;i++)await add(tx,`t${i}`,'ultimate');
|
||||
assert(typeof(await mark(tx,1,true)).done==='number','Tick');assert((await mark(tx,1,false)).done===undefined,'Reopen');
|
||||
for(const todo of (await list(tx)).slice(0,-1))todo.done=todo.id;
|
||||
await mark(tx,DONE_LIMIT+2,true);equal((await list(tx)).map((todo:any)=>todo.id),Array.from({length:DONE_LIMIT},(_,i)=>i+3));await rejects(()=>mark(tx,999,true),'No to-do 999');
|
||||
}
|
||||
{
|
||||
const tx=fakeTx(), content=async()=>(await noteFor(tx,ULTIMATE,0))?.entries[0].model[0].content;
|
||||
assert(await content()===undefined,'Empty note');await add(tx,'Mine','ultimate');await add(tx,'Theirs','user');
|
||||
equal(await content(),"Open to-dos (todo tool: done when finished):\nYours:\n- 1: Mine\nThe user's:\n- 2: Theirs");assert(await content()===undefined,'Unchanged note');
|
||||
await mark(tx,1,true);await mark(tx,2,true);equal(await content(),'No to-dos are open.');assert(await content()===undefined,'Unchanged empty note');
|
||||
}
|
||||
{
|
||||
const tx=fakeTx(), extension=todosModule.default(), tool=extension.tools[0];
|
||||
const run=(conversationId:number,args:any)=>tool.execute(args,{conversationId,commit:(fn:any)=>fn(tx)},context);
|
||||
await rejects(()=>run(9,{action:'list'}),'Only @ultimate');equal(await run(ULTIMATE,{action:'add',text:'Ship it',owner:'user'}),{output:[{type:'text',text:'Added 1 for user: Ship it'}]});
|
||||
equal(await run(ULTIMATE,{action:'list'}),{output:[{type:'text',text:"The user's:\n- 1: Ship it"}]});
|
||||
const before=(conversationId:number)=>extension.hooks[0].handlers.beforeInput({}, {tx,conversationId},context);
|
||||
assert(await before(9)===undefined,'Nonstanding note');assert(await before(ULTIMATE)!==undefined,'Standing note');
|
||||
}
|
||||
// The real world also calls todo through the loaded factory and places its note.
|
||||
await harness.commit(async(tx:any)=>{(await tx.doc(Standing,root.id)).ultimate=root.id;},context);
|
||||
assert(text(await call({action:'add',text:'Real world',owner:'ultimate'},root.id,'todo')).includes('Added 1'),'Loaded todo tool');
|
||||
assert(text(await call({action:'list'},root.id,'todo')).includes('Real world'),'Loaded todo list');
|
||||
assert((await root.context(context)).entries.some((entry:any)=>entry.kind==='todos.note'),'Real beforeInput note');
|
||||
await fs.write('/saved',JSON.stringify({plan:await harness.snapshot(Plan,root.id,context),index:await harness.snapshot(TaskIndex,root.id,context),todos:await harness.snapshot(Todos,root.id,context)}));
|
||||
}
|
||||
} finally {await loader.close();await harness.close(context);}
|
||||
@@ -0,0 +1,96 @@
|
||||
//! The plan and todos ports run through real world extension snapshots and survive restart.
|
||||
use sausage::world::World;
|
||||
use std::path::Path;
|
||||
const BUNDLE: &str = include_str!("../js/plan-todos-test-bundle.js");
|
||||
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn plan_todos_and_restart() -> anyhow::Result<()> {
|
||||
let userland = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland");
|
||||
let target = Path::new(env!("CARGO_TARGET_TMPDIR")).parent().unwrap();
|
||||
if !userland.join("cwd-init/init.o").is_file() {
|
||||
assert!(
|
||||
std::process::Command::new("nu")
|
||||
.args(["--no-config-file", "cwd-init/build.nu"])
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
}
|
||||
assert!(
|
||||
std::process::Command::new("cargo")
|
||||
.args(["build", "--release", "-p", "git", "--manifest-path"])
|
||||
.arg(userland.join("Cargo.toml"))
|
||||
.args(["--target", "wasm32-wasip2", "--target-dir"])
|
||||
.arg(target)
|
||||
.args(["-j", "2", "--locked"])
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
let dir = tempfile::tempdir()?;
|
||||
let disk = dir.path().join("world");
|
||||
let _world = World::open(dir.path())?;
|
||||
std::fs::create_dir_all(disk.join("bin"))?;
|
||||
std::fs::copy(
|
||||
target.join("wasm32-wasip2/release/git.wasm"),
|
||||
disk.join("bin/git.wasm"),
|
||||
)?;
|
||||
let repo = disk.join("src/sausage");
|
||||
std::fs::create_dir_all(&repo)?;
|
||||
assert!(
|
||||
std::process::Command::new("git")
|
||||
.args(["init", "-b", "main"])
|
||||
.arg(&repo)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
for name in ["plan", "todos"] {
|
||||
let destination = repo.join("extensions").join(name);
|
||||
std::fs::create_dir_all(&destination)?;
|
||||
std::fs::copy(
|
||||
Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.join("../extensions")
|
||||
.join(name)
|
||||
.join("server.ts"),
|
||||
destination.join("server.ts"),
|
||||
)?;
|
||||
}
|
||||
assert!(
|
||||
std::process::Command::new("git")
|
||||
.args(["add", "."])
|
||||
.current_dir(&repo)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
assert!(
|
||||
std::process::Command::new("git")
|
||||
.args([
|
||||
"-c",
|
||||
"user.name=Test",
|
||||
"-c",
|
||||
"user.email=test@example.com",
|
||||
"commit",
|
||||
"--no-gpg-sign",
|
||||
"-m",
|
||||
"plan and todos"
|
||||
])
|
||||
.current_dir(&repo)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
for restart in [false, true] {
|
||||
let world = World::open(dir.path())?;
|
||||
let exec = sausage::process::exec(World {
|
||||
root: world.root.clone(),
|
||||
host_dir: world.host_dir.clone(),
|
||||
processes: world.processes.clone(),
|
||||
})?;
|
||||
sausage::harness::run_module(
|
||||
world,
|
||||
exec,
|
||||
&format!("globalThis.planTodosRestart={restart};\n{BUNDLE}"),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user