Skip to content

Commit b45bb16

Browse files
tarastaras
committed
Merge main after workflow supervision delivery
2 parents 0e83223 + 9b98cbe commit b45bb16

6 files changed

Lines changed: 1064 additions & 3 deletions

File tree

‎examples/adversarial-implementation/README.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -366,6 +366,14 @@ the input file.
366366
External reviewer models and publication use their existing workflows. The
367367
phase entry retains the implementation workflow's role and model policy.
368368

369+
## Supervise a live REPL phase
370+
371+
Keep the original ACP call pending, or reconnect passively with the existing
372+
exact-message `--follow` command. [Supervision](supervision.md) explains progress
373+
rendering with `<Query>`, completion evidence and safe detachment. Follow waits
374+
for an exchange; the selected entry and canonical phase or Ticket result still
375+
determine the work's outcome.
376+
369377
## Responsibilities and models
370378

371379
| Owner | Work | Model and effort |

‎examples/adversarial-implementation/progress-reader.node.ts‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
import { execute, inlineSource } from "@executablemd/core";
2+
import { InMemoryStream } from "@executablemd/durable-streams";
3+
import { useHostFiles } from "@executablemd/runtime";
4+
import { readTextFile } from "@effectionx/fs";
5+
import { fromReadable } from "@effectionx/node/stream";
6+
import { main, scoped } from "effection";
7+
import process from "node:process";
8+
import { z } from "zod";
9+
10+
const argumentsSchema = z.tuple([
11+
z.string().min(1),
12+
z
13+
.string()
14+
.regex(/^[1-9][0-9]*$/)
15+
.transform(Number),
16+
]);
17+
const renderedSchema = z.object({ text: z.string(), terminal: z.json().nullable() });
18+
const MAX_LINE = 1024 * 1024;
19+
20+
await main(function* (args) {
21+
const [document, entry] = argumentsSchema.parse(args);
22+
const source = yield* readTextFile(document);
23+
yield* useHostFiles();
24+
const incoming = yield* fromReadable(process.stdin);
25+
const decoder = new TextDecoder("utf-8", { fatal: true });
26+
let pending = "";
27+
let previous: string | undefined;
28+
let terminalEmitted = false;
29+
30+
while (true) {
31+
const next = yield* incoming.next();
32+
pending += next.done ? decoder.decode() : decoder.decode(next.value, { stream: true });
33+
let newline = pending.indexOf("\n");
34+
while (newline >= 0) {
35+
if (newline > MAX_LINE) {
36+
throw new Error("The progress stream exceeded the progress line limit.");
37+
}
38+
const line = pending.slice(0, newline);
39+
pending = pending.slice(newline + 1);
40+
if (line.length > 0) {
41+
const update = z.json().parse(JSON.parse(line));
42+
const rendered = yield* scoped(function* () {
43+
const execution = yield* execute({
44+
...inlineSource(source),
45+
stream: new InMemoryStream(),
46+
includes: [],
47+
props: { update, entry },
48+
});
49+
const result = yield* execution;
50+
if (!result.ok) {
51+
throw result.error;
52+
}
53+
return renderedSchema.parse(result.value);
54+
});
55+
if (rendered.text !== previous) {
56+
process.stderr.write(rendered.text + "\n");
57+
previous = rendered.text;
58+
}
59+
if (rendered.terminal !== null && !terminalEmitted) {
60+
process.stdout.write(JSON.stringify(rendered.terminal) + "\n");
61+
terminalEmitted = true;
62+
}
63+
}
64+
newline = pending.indexOf("\n");
65+
}
66+
if (pending.length > MAX_LINE) {
67+
throw new Error("The progress stream exceeded the progress line limit.");
68+
}
69+
if (next.done) {
70+
if (pending.trim().length > 0) {
71+
throw new Error("The progress stream ended with an incomplete frame.");
72+
}
73+
break;
74+
}
75+
}
76+
});

‎examples/adversarial-implementation/progress.md‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
---
2+
props:
3+
type: object
4+
properties:
5+
update: {type: object}
6+
entry: {type: integer, minimum: 1}
7+
required: [update, entry]
8+
additionalProperties: false
9+
returns:
10+
type: object
11+
---
12+
13+
<Query value={props.update} yq='{"phase": (.activity.phase // "waiting"), "response": (.activity.response // "none"), "replies": (.outputs | length), "questions": ([((.inputs // [])[]) | select(.kind == "question")] | length), "permissions": ([((.inputs // [])[]) | select(.kind == "permission")] | length), "ending": .ending}' as="progress" />
14+
<Let as="text" value={"[entry " + props.entry + "] " + (progress.ending ? "exchange " + progress.ending.outcome : progress.phase) + "; response=" + progress.response + "; replies=" + progress.replies + "; questions=" + progress.questions + "; permissions=" + progress.permissions} />
15+
<Return value={{text,terminal:props.update.ending ? props.update : null}} />

‎examples/adversarial-implementation/supervise.md‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
---
2+
props:
3+
type: object
4+
properties:
5+
step:
6+
type: object
7+
properties:
8+
xmd: {type: array, minItems: 1, items: {type: string, minLength: 1}}
9+
progress: {type: array, minItems: 1, items: {type: string, minLength: 1}}
10+
attachment: {type: string, pattern: '^/'}
11+
message:
12+
type: object
13+
properties:
14+
execution: {type: string, minLength: 1}
15+
conversation: {type: string, minLength: 1}
16+
id: {type: string, minLength: 1}
17+
required: [execution, conversation, id]
18+
additionalProperties: false
19+
entry: {type: integer, minimum: 1}
20+
resultFile: {type: string, pattern: '^/'}
21+
resultSchema: {type: object}
22+
required: [xmd, progress, attachment, message, entry]
23+
additionalProperties: false
24+
required: [step]
25+
additionalProperties: false
26+
returns:
27+
type: object
28+
---
29+
30+
<Let as="step" value={props.step} />
31+
<Let as="followArguments" value={[...step.xmd,"sidekick","read","--attach",step.attachment,"message",encodeURIComponent(JSON.stringify(step.message)),"--follow"]} />
32+
<Let as="followCommand" value={followArguments.map(value => "'" + value.replaceAll("'","'\"'\"'") + "'").join(" ")} />
33+
<Let as="progressCommand" value={[...step.progress,String(step.entry)].map(value => "'" + value.replaceAll("'","'\"'\"'") + "'").join(" ")} />
34+
35+
<Let as="followedText">
36+
```bash exec
37+
set -o pipefail
38+
{followCommand} | {progressCommand}
39+
follow_status=$?
40+
if [ "$follow_status" -ne 0 ]; then
41+
printf '{"followExitCode":%s}\n' "$follow_status"
42+
fi
43+
```
44+
</Let>
45+
46+
<Let as="lastLine" value={followedText.trim().split("\n").at(-1) ?? ""} />
47+
<SafeParse schema={{type:"object",properties:{message:{const:step.message},ending:{type:"object",properties:{joined:{const:true},outcome:{enum:["completed","failed","cancelled"]}},required:["joined","outcome"]}},required:["message","ending"]}} as="exchange">{lastLine}</SafeParse>
48+
<Let as="report" value={{entry:step.entry,status:"unavailable",joinedExchange:false,diagnostic:"Follow failed or returned no validated joined receipt. Inspect stderr and reconcile the original call; no work was resent.",resultStatus:"unavailable",result:null}} />
49+
<If condition={exchange.ok}>
50+
<Let as="report" value={{...report,joinedExchange:true,diagnostic:"The attachment could not be copied for the selected message."}} />
51+
<TempDir as="attachmentDirectory" />
52+
<Let as="privateAttachment" value={attachmentDirectory + "/attachment.json"} />
53+
<Let as="copyCommand" value={["install","-m","600",step.attachment,privateAttachment].map(value => "'" + value.replaceAll("'","'\"'\"'") + "'").join(" ")} />
54+
```bash exec as="attachmentCopy"
55+
exec {copyCommand}
56+
```
57+
<If condition={attachmentCopy.exitCode === 0}>
58+
<Let as="descriptorCommand" value={"cat -- '" + privateAttachment.replaceAll("'","'\"'\"'") + "'"} />
59+
```bash exec as="descriptorRead"
60+
exec {descriptorCommand}
61+
```
62+
<Let as="report" value={{...report,diagnostic:"The copied attachment did not validate against the selected execution and conversation."}} />
63+
<If condition={descriptorRead.exitCode === 0}>
64+
<SafeParse schema={{type:"object",properties:{execution:{const:step.message.execution},conversation:{const:step.message.conversation}},required:["execution","conversation"]}} as="attachmentEvidence">{descriptorRead.stdout}</SafeParse>
65+
<If condition={attachmentEvidence.ok}>
66+
<Let as="entryArguments" value={[...step.xmd,"sidekick","read","--attach",privateAttachment,"entry",String(step.entry),"journal","--yq",'{"entry": .entry, "status": .status, "failure": .failure}']} />
67+
<Let as="entryCommand" value={entryArguments.map(value => "'" + value.replaceAll("'","'\"'\"'") + "'").join(" ")} />
68+
```bash exec as="entryRead"
69+
exec {entryCommand}
70+
```
71+
<Let as="report" value={{...report,diagnostic:entryRead.exitCode === 0 ? "Entry read did not validate a complete result for the selected entry." : entryRead.stderr}} />
72+
<If condition={entryRead.exitCode === 0}>
73+
<SafeParse schema={{type:"object",properties:{value:{type:"object",properties:{entry:{const:step.entry},status:{enum:["completed","failed","cancelled","running","unfinished"]},failure:{type:["string","null"]}},required:["entry","status","failure"],additionalProperties:false},next:{type:"null"},prev:{type:"null"},last:{type:"null"}},required:["value","next","prev","last"],additionalProperties:false}} as="entryEvidence">{entryRead.stdout}</SafeParse>
74+
<If condition={entryEvidence.ok}>
75+
<Let as="page" value={entryEvidence.value.value} />
76+
<Let as="report" value={{entry:step.entry,status:page.status,exchangeOutcome:exchange.value.ending.outcome,joinedExchange:true,diagnostic:page.failure,read:{entry:step.entry,view:"journal"},resultStatus:"not-requested",result:null}} />
77+
<If condition={page.status === "completed" && !!step.resultFile}>
78+
<Let as="resultCommand" value={"cat -- '" + step.resultFile.replaceAll("'","'\"'\"'") + "'"} />
79+
```bash exec as="resultRead"
80+
exec {resultCommand}
81+
```
82+
<Let as="report" value={{...report,status:"unavailable",entryStatus:page.status,resultStatus:"unavailable",diagnostic:resultRead.stderr}} />
83+
<If condition={resultRead.exitCode === 0}>
84+
<SafeParse schema={step.resultSchema ?? {type:"object"}} as="canonical">{resultRead.stdout}</SafeParse>
85+
<If condition={canonical.ok}>
86+
<Let as="report" value={{...report,status:page.status,resultStatus:"readable",diagnostic:page.failure,result:canonical.value}} />
87+
<Else>
88+
<Let as="report" value={{...report,resultStatus:"invalid",diagnostic:"The selected canonical result did not validate. Preserve its attempt and reconcile the original work.",validationErrors:canonical.errors}} />
89+
</Else>
90+
</If>
91+
</If>
92+
</If>
93+
</If>
94+
</If>
95+
</If>
96+
</If>
97+
</If>
98+
</If>
99+
<Return value={report} />

‎examples/adversarial-implementation/supervision.md‎

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
# Supervise a live REPL phase
2+
3+
Read this when a workflow phase runs inside a retained REPL and its coordinator
4+
needs progress and completion. Keep one wait open instead of repeatedly reading
5+
result-file existence, timestamps or log tails.
6+
7+
## Applies to
8+
9+
Coordinators observing an existing live REPL. These commands run outside its
10+
Sidekick conversation; they submit no work and call no agent. Keep the owning
11+
parent alive and retain the original child, role conversations and counters.
12+
13+
## Instructions
14+
15+
1. Keep the original ACP invocation pending when it already waits for the work.
16+
Its acknowledgement supplies the exact message reference needed to reconnect.
17+
2. If observation must reconnect, use that acknowledged message's existing
18+
`--follow` command. Preserve its execution, conversation and message identity;
19+
do not search for whichever message is latest.
20+
3. Receive the stream's updates while that single command waits. After its
21+
validated joined ending, use the supervisor below to read the associated
22+
entry's Journal once and inspect the original canonical phase or Ticket result.
23+
Retain the launch association between that message, entry and reserved result
24+
path; a matching entry number alone does not establish that association.
25+
4. Preserve incomplete or uncertain outcomes. A disconnected owner, missing
26+
receipt or invalid result calls for reconciliation of the original work,
27+
never an automatic resend or a new attempt.
28+
29+
An exchange that submits and waits for an entry covers that entry's completion
30+
and cleanup. An independently running entry can remain active after an unrelated
31+
exchange ends. Entry reads are snapshots; `entry ... --follow` is unsupported.
32+
33+
## Example
34+
35+
Replace the attachment path and all three reference fields with the original
36+
acknowledged values. The example identities are illustrative, not live sessions:
37+
38+
```bash
39+
attachment=/absolute/private/attachment.json
40+
reference='{"execution":"saved-execution","conversation":"saved-conversation","id":"saved-message"}'
41+
xmd sidekick read --attach "$attachment" message "$reference" --follow
42+
```
43+
44+
The command emits JSON snapshots as the exchange changes and returns its joined
45+
ending. It does not issue status polls or prompt the agent. Once it finishes,
46+
inspect the entry recorded for that submission, replacing `14` with its actual
47+
number:
48+
49+
```bash
50+
xmd sidekick read --attach "$attachment" entry 14 journal
51+
```
52+
53+
A completed exchange can contain a failed or cancelled entry. A completed entry
54+
can return a Ticket on hold. Neither process exit nor exchange completion
55+
replaces the workflow's own verdict or delivery checks.
56+
57+
### Format a received snapshot with Query
58+
59+
Use built-in `<Query>` to select progress from a snapshot already received. This
60+
complete document accepts one snapshot through the aggregate `--props` JSON
61+
argument and renders a compact summary without copying provisional reply text:
62+
63+
```xmd
64+
---
65+
props:
66+
type: object
67+
properties:
68+
snapshot: {type: object}
69+
required: [snapshot]
70+
additionalProperties: false
71+
---
72+
73+
<Query value={props.snapshot}
74+
yq='{"phase": (.activity.phase // "waiting"), "response": .activity.response, "replies": (.outputs | length), "ending": .ending}'
75+
as="progress" />
76+
<Output><Json value={progress} /></Output>
77+
```
78+
79+
Query selects from the supplied value; it does not subscribe, read another
80+
snapshot or wait for completion. For live formatted progress, keep one passive
81+
reader feeding received snapshots into the rendering document. Emit a summary
82+
when it changes and retain the exact terminal receipt separately. This avoids
83+
starting another XMD process for every token update. The standalone Query
84+
example above formats one value; it is not itself a streaming command.
85+
86+
### Run the supervisor example
87+
88+
[supervise.md](supervise.md) composes the existing follow command with
89+
[progress.md](progress.md) and its [persistent reader](progress-reader.node.ts).
90+
Query selects the progress fields. The reader feeds snapshots to that document,
91+
prints a changed summary on stderr and forwards the original terminal receipt
92+
once on stdout. After validating that receipt, the supervisor copies the original
93+
attachment once into an invocation-private directory with a mode-0600 file. It
94+
checks that copy's execution and conversation against the acknowledged message,
95+
then uses the unchanged copy for the one Journal connection. The live handshake
96+
validates the whole descriptor; copied bytes alone prove no live owner. Replacing
97+
the original filename after copying cannot retarget that read. The directory is
98+
removed when the supervisor ends, including failure or cancellation.
99+
100+
The current Journal response is `{value,next,prev,last}`. The supervisor selects
101+
`{"entry": .entry, "status": .status, "failure": .failure}` with `--yq`, validates those
102+
facts under `value`, and requires null navigation so a partial failure diagnostic
103+
cannot stand in for a complete result. The Journal body supplies no execution or
104+
conversation identity; that association comes from the validated descriptor and
105+
its successful handshake on the same connection. It owns no agent or REPL.
106+
107+
```text
108+
Exact message follow -> one persistent Query renderer -> compact live stderr
109+
-> validated joined receipt
110+
-> private descriptor copy + live authenticated Journal read
111+
-> completed entry permits one canonical-result read -> JSON
112+
```
113+
114+
Save this input under a new private coordination filename. Replace paths and
115+
identities with the original launch values; the examples are illustrative:
116+
117+
```json
118+
{
119+
"step": {
120+
"xmd": ["deno", "run", "--quiet", "--allow-all", "/absolute/workflow/packages/cli/src/deno.ts"],
121+
"progress": [
122+
"deno", "run", "--quiet", "--allow-all",
123+
"/absolute/workflow/examples/adversarial-implementation/progress-reader.node.ts",
124+
"/absolute/workflow/examples/adversarial-implementation/progress.md"
125+
],
126+
"attachment": "/absolute/private/attachment.json",
127+
"message": {"execution": "saved-execution", "conversation": "saved-conversation", "id": "saved-message"},
128+
"entry": 14,
129+
"resultFile": "/absolute/work/attempt/result.json",
130+
"resultSchema": {"type": "object"}
131+
}
132+
}
133+
```
134+
135+
Use the selected CLI command in `xmd` and the existing runtime launcher in
136+
`progress`. The reader also runs with Node/tsx or Bun; it uses their Node stream
137+
interfaces. Pass the canonical result's existing schema when available. The
138+
default object schema establishes JSON readability, not workflow acceptance.
139+
Then keep this invocation pending until it returns:
140+
141+
```bash
142+
xmd run /absolute/workflow/examples/adversarial-implementation/supervise.md \
143+
--props "$(cat -- /absolute/private/supervision-input.json)" --raw
144+
```
145+
146+
The root `step` object uses aggregate `--props`, not `--props-step`. The result
147+
keeps `status` for the entry and `exchangeOutcome` for its exchange. A readable
148+
canonical artifact appears unchanged under `result`, with `resultStatus=readable`.
149+
Without `resultFile`, it reports `not-requested`. A missing or invalid requested
150+
result reports `unavailable`, preserving `entryStatus` separately.
151+
152+
Copy, tuple-validation, read or connection failure leaves entry evidence
153+
unavailable and skips the canonical-result read. A previously validated joined
154+
receipt still reports `joinedExchange=true`; it does not establish entry
155+
completion. A mismatched tuple is rejected before Journal, while owner loss or
156+
conversation change fails the live handshake. An oversized selected diagnostic
157+
may refuse under the Journal bound and remains unavailable evidence. The
158+
supervisor does not follow pages, poll, resend work or convert a canonical hold
159+
into a passing verdict.
160+
161+
An enclosing `<Let as="receipt">` can capture a foreground command's stdout
162+
while stderr forwards live progress. A bound `exec as="receipt"` captures both
163+
channels and displays neither. Keep the final structured result on stdout and
164+
compact progress on stderr when composing such a supervisor. A missing or
165+
invalid terminal receipt remains incomplete evidence even if the process exits.
166+
167+
## Common mistake
168+
169+
Resending an assignment because its observer disconnected can duplicate effects.
170+
Detaching a passive follow releases observation and leaves the producer running;
171+
explicit ACP cancellation is a different operation. Reconnect to the original
172+
acknowledged exchange and reconcile its result without changing session identity,
173+
consumed correction limits or reserved attempts.
174+
175+
## Read next
176+
177+
Read [Run a separate phase or continue a saved ticket](README.md#run-a-separate-phase-or-continue-a-saved-ticket)
178+
for `<Phase.Run>` and `<Ticket.Continue>`. The language specification defines
179+
[Query](../../specs/executable-mdx-spec.md#622-selecting-part-of-a-value-query)
180+
and [foreground command channels](../../specs/executable-mdx-spec.md#36-foreground-execution-capture-and-retention).

0 commit comments

Comments
 (0)