Skip to content

Commit 9af8a3b

Browse files
ZhiXiao-Linclaude
andauthored
docs(orchestration): grammar README section + SDK examples (Phase 5e) (#58)
Documents the programmable orchestration grammar (parallel / pipeline / resumable) for both SDKs with side-by-side Python/Node snippets, including the step-spec shape, schema'd steps, the no-inter-stage-barrier pipeline semantics, and the pipeline-stage no-throw constraint. Adds runnable examples: - sdk/python/examples/orchestration_workflow.py - sdk/node/examples/orchestration/parallel-pipeline.mjs Both load .a3s/config.acl at runtime and demonstrate parallel + pipeline; syntax-checked (py_compile / node --check). Running them requires the built native addon (maturin / napi) — they double as the SDK-level validation reference for the callback bridges. Completes Workflow Phase 5 (full grammar, both SDKs). Co-authored-by: Claude <claude@anthropic.com>
1 parent a21037f commit 9af8a3b

3 files changed

Lines changed: 213 additions & 0 deletions

File tree

README.md

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -571,6 +571,69 @@ session2.setBudgetGuard({
571571

572572
---
573573

574+
## Programmable Orchestration
575+
576+
Beyond *model-driven* delegation (the agent calling `task`/`parallel_task`), a
577+
session exposes a **deterministic, programmable** orchestration grammar: you
578+
decide the fan-out, chaining, and resume in code. Steps run through the
579+
session's `AgentExecutor`; a host (书安OS) can substitute its own executor to
580+
place steps across a cluster — the grammar is identical either way.
581+
582+
A **step** is `{ task_id, agent, description, prompt, max_steps?,
583+
parent_session_id?, output_schema? }`. With `output_schema`, the step returns a
584+
schema-validated object in `structured`.
585+
586+
```python
587+
# Fan out N steps; results come back in input order. A failed step is
588+
# success=False, not a thrown error.
589+
outcomes = session.parallel([
590+
{"task_id": "a", "agent": "explore", "description": "find auth", "prompt": "Where is auth handled?"},
591+
{"task_id": "b", "agent": "review", "description": "review", "prompt": "Review src/auth.rs",
592+
"output_schema": {"type": "object", "properties": {"verdict": {"type": "string"}}, "required": ["verdict"]}},
593+
])
594+
print(outcomes[1]["structured"]) # {"verdict": "..."}
595+
596+
# Pipeline: each item flows through stages with NO barrier between them —
597+
# item A can be in stage 2 while item B is still in stage 1. A stage returns
598+
# the next step (deriving from the previous outcome) or None to stop.
599+
results = session.pipeline(
600+
["src/auth.rs", "src/db.rs"],
601+
[
602+
lambda ctx: {"task_id": "rev", "agent": "review", "description": "review", "prompt": f"Review {ctx['item']}"},
603+
lambda ctx: {"task_id": "fix", "agent": "general", "description": "verify",
604+
"prompt": f"Verify this review: {ctx['previous']['output']}"},
605+
],
606+
)
607+
608+
# Resumable: progress is journaled under a workflow id via the session store,
609+
# so an interrupted run (or one resumed on another node) skips completed steps.
610+
outcomes = session.parallel_resumable(specs, "nightly-audit")
611+
```
612+
613+
```javascript
614+
// Node — identical shapes, camelCase. Promises resolve to the same outcomes.
615+
const outcomes = await session.parallel([
616+
{ taskId: "a", agent: "explore", description: "find auth", prompt: "Where is auth handled?" },
617+
{ taskId: "b", agent: "review", description: "review", prompt: "Review src/auth.rs" },
618+
]);
619+
620+
const results = await session.pipeline(
621+
["src/auth.rs", "src/db.rs"],
622+
[
623+
(ctx) => ({ taskId: "rev", agent: "review", description: "review", prompt: `Review ${ctx.item}` }),
624+
(ctx) => ({ taskId: "fix", agent: "general", description: "verify",
625+
prompt: `Verify this review: ${ctx.previous.output}` }),
626+
],
627+
);
628+
629+
await session.parallelResumable(specs, "nightly-audit");
630+
// NOTE: pipeline stage callbacks MUST NOT throw — return null to stop a chain.
631+
// A throw aborts the process (same constraint as setBudgetGuard). A stage that
632+
// hangs past timeoutMs (default 30s) fails closed (treated as null).
633+
```
634+
635+
---
636+
574637
## Design Principles
575638

576639
### 1. Small Kernel
Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
/**
2+
* Programmable orchestration via the A3S Code Node SDK.
3+
*
4+
* session.parallel(specs) — fan out steps, results in input order
5+
* session.pipeline(items, stages) — per-item chains, no inter-stage barrier
6+
* session.parallelResumable(specs, id) — journaled, resumable fan-out
7+
*
8+
* Config is read from .a3s/config.acl at runtime (never hardcoded).
9+
* Run: node sdk/node/examples/orchestration/parallel-pipeline.mjs
10+
*/
11+
import { createRequire } from 'node:module';
12+
import { existsSync } from 'node:fs';
13+
import { dirname, join } from 'node:path';
14+
import { fileURLToPath } from 'node:url';
15+
16+
const require = createRequire(import.meta.url);
17+
const { Agent } = require('@a3s-lab/code');
18+
19+
function repoConfigPath() {
20+
let dir = dirname(fileURLToPath(import.meta.url));
21+
for (let i = 0; i < 10; i++) {
22+
const candidate = join(dir, '.a3s', 'config.acl');
23+
if (existsSync(candidate)) return candidate;
24+
dir = dirname(dir);
25+
}
26+
throw new Error('could not locate .a3s/config.acl above this example');
27+
}
28+
29+
async function main() {
30+
const agent = await Agent.create(repoConfigPath());
31+
const session = agent.session('.', {});
32+
33+
// 1. parallel — independent steps; outcomes in input order.
34+
const outcomes = await session.parallel([
35+
{ taskId: 'langs', agent: 'general', description: 'list', prompt: 'Name three systems languages.', maxSteps: 2 },
36+
{ taskId: 'safe', agent: 'general', description: 'classify', prompt: 'Is Rust memory-safe without a GC? yes/no.', maxSteps: 2 },
37+
]);
38+
for (const o of outcomes) console.log(`[parallel] ${o.taskId}: success=${o.success}`);
39+
40+
// 2. pipeline — stage 2 builds on stage 1's output. Return null to stop a
41+
// chain. A stage callback MUST NOT throw (return null on error).
42+
const results = await session.pipeline(
43+
['the Rust programming language'],
44+
[
45+
(ctx) => ({ taskId: 'sum', agent: 'general', description: 'summarize', prompt: `In one sentence, what is ${ctx.item}?`, maxSteps: 2 }),
46+
(ctx) => ({ taskId: 'cls', agent: 'general', description: 'classify',
47+
prompt: `Reply YES or NO: does this describe a programming language?\n\n${ctx.previous.output}`, maxSteps: 2 }),
48+
],
49+
);
50+
for (const r of results) console.log(`[pipeline] final=${r === null ? null : JSON.stringify(r.output.slice(0, 60))}`);
51+
}
52+
53+
main().catch((err) => {
54+
console.error(err);
55+
process.exit(1);
56+
});
Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
#!/usr/bin/env python3
2+
"""
3+
Programmable orchestration via the A3S Code Python SDK.
4+
5+
Demonstrates the deterministic grammar exposed on a session:
6+
- session.parallel(specs) — fan out steps, results in input order
7+
- session.pipeline(items, stages) — per-item chains, no inter-stage barrier
8+
- session.parallel_resumable(...) — journaled, resumable fan-out
9+
10+
Config is loaded from .a3s/config.acl at runtime (never hardcoded).
11+
12+
Run: cd crates/code/sdk/python && python examples/orchestration_workflow.py
13+
"""
14+
15+
import sys
16+
from pathlib import Path
17+
18+
from a3s_code import Agent, SessionOptions
19+
20+
21+
def repo_config_path() -> Path:
22+
"""Find .a3s/config.acl walking up from this file."""
23+
here = Path(__file__).resolve()
24+
for parent in here.parents:
25+
candidate = parent / ".a3s" / "config.acl"
26+
if candidate.exists():
27+
return candidate
28+
raise SystemExit("could not locate .a3s/config.acl above this example")
29+
30+
31+
def main() -> int:
32+
config = str(repo_config_path())
33+
agent = Agent.create(config)
34+
session = agent.session(".", SessionOptions())
35+
36+
# 1. parallel — fan out independent steps; outcomes come back in order.
37+
outcomes = session.parallel(
38+
[
39+
{
40+
"task_id": "langs",
41+
"agent": "general",
42+
"description": "list languages",
43+
"prompt": "Name three systems programming languages, comma-separated.",
44+
"max_steps": 2,
45+
},
46+
{
47+
"task_id": "verdict",
48+
"agent": "general",
49+
"description": "classify",
50+
"prompt": "Is Rust memory-safe without a GC? Answer yes or no.",
51+
"max_steps": 2,
52+
# Schema-validated structured output for this step.
53+
"output_schema": {
54+
"type": "object",
55+
"properties": {"memory_safe": {"type": "boolean"}},
56+
"required": ["memory_safe"],
57+
},
58+
},
59+
]
60+
)
61+
for o in outcomes:
62+
print(f"[parallel] {o['task_id']}: success={o['success']} structured={o.get('structured')}")
63+
64+
# 2. pipeline — each item chains through stages; stage 2 builds on stage 1.
65+
# Return None from a stage (or raise — caught and treated as None) to
66+
# stop that item's chain.
67+
results = session.pipeline(
68+
["the Rust programming language"],
69+
[
70+
lambda ctx: {
71+
"task_id": "summarize",
72+
"agent": "general",
73+
"description": "summarize",
74+
"prompt": f"In one sentence, what is {ctx['item']}?",
75+
"max_steps": 2,
76+
},
77+
lambda ctx: {
78+
"task_id": "classify",
79+
"agent": "general",
80+
"description": "classify",
81+
"prompt": "Reply with one word YES or NO: does this describe a "
82+
f"programming language?\n\n{ctx['previous']['output']}",
83+
"max_steps": 2,
84+
},
85+
],
86+
)
87+
for r in results:
88+
print(f"[pipeline] final={None if r is None else r['output'][:60]!r}")
89+
90+
return 0
91+
92+
93+
if __name__ == "__main__":
94+
sys.exit(main())

0 commit comments

Comments
 (0)