diff --git a/go.work b/go.work index 31ddca69..f2db1a94 100644 --- a/go.work +++ b/go.work @@ -12,5 +12,6 @@ use ( ./labs/18-settle/settle ./labs/20-deltawire ./labs/21-interlock/interlock + ./labs/22-yield/yield ./labs/21-interlock/integrations/pitot ) diff --git a/labs/22-yield/README.md b/labs/22-yield/README.md new file mode 100644 index 00000000..1dd98a2c --- /dev/null +++ b/labs/22-yield/README.md @@ -0,0 +1,71 @@ +# Lab 22 — Yield + +Programmable skills for coding agents. **Turn `SKILL.md` workflows into +resumable programs.** + +> The skill yields the next typed operation. The coding agent performs it +> and resumes the skill. + +A skill keeps its thin `SKILL.md` (so it works wherever skills work today) +and moves its control flow — order, branching, retries, approval, state, +completion — into a deterministic Go program. The model keeps reasoning, +exploration, editing, and judgment. No daemon, no new agent loop, no host +integration: the agent only runs a CLI and follows envelopes. + +## Execution model + +Deterministic re-execution (no resumable coroutines needed): on every +run/resume, `yskill` re-executes the skill program from the top, feeding +recorded responses back in order. At the first unanswered operation the +SDK emits a `yield.v1` request envelope and exits. A replayed step that +produces a different operation than the journal recorded is a divergence +and fails the run loudly — it never silently forks. + +Two processes: + +- **`yskill` (supervisor)** — owns the append-only run log + (`.yield/runs/.jsonl`), sequence and digest binding, response + validation, and every refusal. `run_command` operations are executed by + yskill itself, so command results enter the log as observed fact rather + than the agent's transcription. +- **skill program (subprocess)** — an ordinary Go `main` using + `yield/sdk/yield`; every side effect crosses a yielded primitive + (`AskUser`, `AgentTask`, `RunCommand`, `Require`, `Complete`, plus the + `Blocked`/`Refused` terminals). + +## Layout + +``` +yield/ +├── cmd/yskill/ init · run · resume · inspect · replay · test +├── internal/protocol/ yield.v1 envelopes, digests, schema validation +├── internal/runlog/ append-only JSONL event log +├── internal/guard/ every refusal: stale/dup/wrong-run/schema/digest/unproven +├── internal/engine/ supervisor: subprocess execution, command evidence, terminals +├── sdk/yield/ the skill-program SDK (deterministic replay context) +├── examples/investigate the reference skill (bounded hypothesis loop) +└── docs/locus* certified lifecycle models + verdicts +``` + +## Try it + +``` +go build -o /tmp/yskill ./yield/cmd/yskill +/tmp/yskill test yield/examples/investigate # scripted fixture run to completion +/tmp/yskill run yield/examples/investigate # prints the first operation envelope +``` + +## Formal grounding + +The run lifecycle was designed against Locus verdicts +(`yield/docs/locus-yield.md`): protocol integrity is supervisable +(controllable, no violations), every run reaches an honest terminal +(nonblocking — the `--accept-new-digest` migrate verb is load-bearing), +and portable mode is provably non-diagnosable for off-protocol agent +action — which is stated as a non-guarantee rather than papered over. A +correlated host adapter is the certified post-V1 answer to that gap. + +Out of V1 (deliberately): prose-to-program compilation, a hosted runtime, +a workflow DSL, a marketplace, a multi-agent orchestrator, a security +sandbox, any canonical IR. Working name `Yield`/`yskill` pending a +collision scan before any public projection. diff --git a/labs/22-yield/yield/.gitignore b/labs/22-yield/yield/.gitignore new file mode 100644 index 00000000..249dbc7d --- /dev/null +++ b/labs/22-yield/yield/.gitignore @@ -0,0 +1 @@ +.yield/ diff --git a/labs/22-yield/yield/cmd/yskill/main.go b/labs/22-yield/yield/cmd/yskill/main.go new file mode 100644 index 00000000..c7aca343 --- /dev/null +++ b/labs/22-yield/yield/cmd/yskill/main.go @@ -0,0 +1,329 @@ +// yskill is the supervisor CLI for Yield: it starts runs, validates and +// accepts responses, executes commands as observed fact, and owns the +// append-only run log. The coding agent drives it through six verbs. +package main + +import ( + "encoding/json" + "flag" + "fmt" + "os" + "path/filepath" + + "yield/internal/engine" + "yield/internal/protocol" +) + +const usage = `yskill — turn SKILL.md workflows into resumable programs + +Usage: + yskill init scaffold a skill (or wrap an existing prose skill) + yskill run [--input file] start a run; prints the first operation envelope + yskill resume --response file feed a response; prints the next operation + [--skill dir] [--accept-new-digest] + yskill inspect [--skill dir] print the run's event log + yskill replay [--skill dir] re-derive the run from its log; verify determinism + yskill test run the skill against fixtures/responses.json +` + +func main() { + if len(os.Args) < 2 { + fmt.Fprint(os.Stderr, usage) + os.Exit(2) + } + var err error + switch os.Args[1] { + case "init": + err = cmdInit(os.Args[2:]) + case "run": + err = cmdRun(os.Args[2:]) + case "resume": + err = cmdResume(os.Args[2:]) + case "inspect": + err = cmdInspect(os.Args[2:]) + case "replay": + err = cmdReplay(os.Args[2:]) + case "test": + err = cmdTest(os.Args[2:]) + case "help", "-h", "--help": + fmt.Print(usage) + default: + fmt.Fprintf(os.Stderr, "yskill: unknown verb %q\n\n%s", os.Args[1], usage) + os.Exit(2) + } + if err != nil { + fmt.Fprintf(os.Stderr, "yskill: %v\n", err) + os.Exit(1) + } +} + +func cmdRun(args []string) error { + fs := flag.NewFlagSet("run", flag.ExitOnError) + input := fs.String("input", "", "path to a JSON input file") + if err := fs.Parse(args); err != nil { + return err + } + if fs.NArg() != 1 { + return fmt.Errorf("run takes exactly one skill directory") + } + e, err := engine.New(fs.Arg(0)) + if err != nil { + return err + } + var in json.RawMessage + if *input != "" { + b, err := os.ReadFile(*input) + if err != nil { + return err + } + in = b + } + p, err := e.StartRun(in) + if err != nil { + return err + } + return printProgress(p) +} + +func cmdResume(args []string) error { + fs := flag.NewFlagSet("resume", flag.ExitOnError) + response := fs.String("response", "", "path to the response envelope JSON") + skillDir := fs.String("skill", ".", "skill directory the run belongs to") + migrate := fs.Bool("accept-new-digest", false, "explicitly rebind the run to the current skill source digest") + if err := fs.Parse(args); err != nil { + return err + } + if fs.NArg() != 1 || *response == "" { + return fmt.Errorf("resume takes a run id and --response file") + } + e, err := engine.New(*skillDir) + if err != nil { + return err + } + b, err := os.ReadFile(*response) + if err != nil { + return err + } + p, err := e.Resume(fs.Arg(0), b, *migrate) + if err != nil { + return err + } + return printProgress(p) +} + +func cmdInspect(args []string) error { + fs := flag.NewFlagSet("inspect", flag.ExitOnError) + skillDir := fs.String("skill", ".", "skill directory the run belongs to") + if err := fs.Parse(args); err != nil { + return err + } + e, err := engine.New(*skillDir) + if err != nil { + return err + } + if fs.NArg() == 0 { + ids, err := e.ListRuns() + if err != nil { + return err + } + for _, id := range ids { + fmt.Println(id) + } + return nil + } + l, err := e.Log(fs.Arg(0)) + if err != nil { + return err + } + for _, ev := range l.Events() { + fmt.Printf("%03d %-22s %s %s\n", ev.Seq, ev.Type, ev.At.Format("2006-01-02T15:04:05Z"), compact(ev.Data)) + } + return nil +} + +func cmdReplay(args []string) error { + fs := flag.NewFlagSet("replay", flag.ExitOnError) + skillDir := fs.String("skill", ".", "skill directory the run belongs to") + if err := fs.Parse(args); err != nil { + return err + } + if fs.NArg() != 1 { + return fmt.Errorf("replay takes exactly one run id") + } + e, err := engine.New(*skillDir) + if err != nil { + return err + } + p, err := e.Replay(fs.Arg(0)) + if err != nil { + return err + } + fmt.Println("replay: deterministic — the log reproduces the run's frontier") + return printProgress(p) +} + +// cmdTest drives a skill against fixtures/responses.json: an ordered map +// of request id -> scripted result. run_command operations execute for +// real; everything else is answered from the script. +func cmdTest(args []string) error { + fs := flag.NewFlagSet("test", flag.ExitOnError) + if err := fs.Parse(args); err != nil { + return err + } + if fs.NArg() != 1 { + return fmt.Errorf("test takes exactly one skill directory") + } + dir := fs.Arg(0) + b, err := os.ReadFile(filepath.Join(dir, "fixtures", "responses.json")) + if err != nil { + return fmt.Errorf("fixtures/responses.json is required: %w", err) + } + var script map[string]json.RawMessage + if err := json.Unmarshal(b, &script); err != nil { + return fmt.Errorf("fixtures/responses.json does not decode: %w", err) + } + e, err := engine.New(dir) + if err != nil { + return err + } + p, err := e.StartRun(nil) + if err != nil { + return err + } + for p.Terminal == nil { + result, ok := script[p.Envelope.Request.ID] + if !ok { + return fmt.Errorf("no scripted response for request %q (sequence %d)", p.Envelope.Request.ID, p.Envelope.Sequence) + } + resp, err := json.Marshal(protocol.ResponseEnvelope{ + RunID: p.RunID, Sequence: p.Envelope.Sequence, + RequestID: p.Envelope.Request.ID, Status: "completed", Result: result, + }) + if err != nil { + return err + } + if p, err = e.Resume(p.RunID, resp, false); err != nil { + return err + } + } + fmt.Printf("test: run %s reached %s\n", p.RunID, p.Terminal.Status) + if p.Terminal.Status != protocol.StatusCompleted { + return fmt.Errorf("terminal status %s: %s", p.Terminal.Status, p.Terminal.Reason) + } + return nil +} + +func printProgress(p *engine.Progress) error { + if p.Terminal != nil { + fmt.Printf("run %s: %s\n", p.RunID, p.Terminal.Status) + if p.Terminal.Reason != "" { + fmt.Printf("reason: %s\n", p.Terminal.Reason) + } + if len(p.Terminal.Result) > 0 { + fmt.Printf("result: %s\n", compact(p.Terminal.Result)) + } + return nil + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + return enc.Encode(p.Envelope) +} + +func compact(raw json.RawMessage) string { + if len(raw) == 0 { + return "" + } + s := string(raw) + if len(s) > 140 { + s = s[:140] + "…" + } + return s +} + +// cmdInit scaffolds a minimal skill: a thin SKILL.md, a skill program, and +// a fixtures directory. If the directory already has a SKILL.md, it is +// preserved — init wraps existing prose skills rather than replacing them. +func cmdInit(args []string) error { + fs := flag.NewFlagSet("init", flag.ExitOnError) + sdkPath := fs.String("sdk", "", "filesystem path to the yield module (written as a go.mod replace directive)") + if err := fs.Parse(args); err != nil { + return err + } + if fs.NArg() != 1 { + return fmt.Errorf("init takes exactly one directory") + } + dir := fs.Arg(0) + name := filepath.Base(dir) + if err := os.MkdirAll(filepath.Join(dir, "fixtures"), 0o755); err != nil { + return err + } + writeIfAbsent := func(rel, content string) error { + path := filepath.Join(dir, rel) + if _, err := os.Stat(path); err == nil { + fmt.Printf("init: %s exists, preserved\n", rel) + return nil + } + return os.WriteFile(path, []byte(content), 0o644) + } + if err := writeIfAbsent("SKILL.md", fmt.Sprintf(skillMD, name)); err != nil { + return err + } + if err := writeIfAbsent("main.go", mainGo); err != nil { + return err + } + if err := writeIfAbsent("fixtures/responses.json", "{\n \"confirm-start\": {\"value\": \"yes\"}\n}\n"); err != nil { + return err + } + gomod := fmt.Sprintf("module %s\n\ngo 1.26.5\n\nrequire yield v0.0.0\n", name) + if *sdkPath != "" { + gomod += fmt.Sprintf("\nreplace yield => %s\n", *sdkPath) + } else { + gomod += "\n// Point this at your yield checkout:\n// replace yield => ../path/to/yield\n" + } + if err := writeIfAbsent("go.mod", gomod); err != nil { + return err + } + fmt.Printf("init: skill %q scaffolded in %s\n", name, dir) + return nil +} + +const skillMD = `--- +name: %s +description: TODO — one line on what this skill does. +--- + +Run: + + yskill run . + +Follow each returned operation exactly. + +- ` + "`ask_user`" + `: ask the user using the host's normal interface. +- ` + "`agent_task`" + `: perform the task and return schema-valid JSON. +- ` + "`run_command`" + `: yskill executes it itself; you will not see this kind. + +Resume the run after each operation: + + yskill resume --response response.json + +Do not skip an operation or invent its response. +` + +const mainGo = `package main + +import ( + "yield/sdk/yield" +) + +func main() { + yield.Main(func(ctx *yield.Context) (yield.Outcome, error) { + answer := ctx.AskUser("confirm-start", "Ready to start?") + if answer != "yes" { + return yield.Outcome{}, ctx.Refused("user declined to start") + } + tests := ctx.RunCommand("run-tests", "true", 60) + ctx.Require(tests.ExitCode == 0, "the test command passes", tests) + return ctx.Complete(map[string]string{"status": "ok"}) + }) +} +` diff --git a/labs/22-yield/yield/docs/locus-yield.md b/labs/22-yield/yield/docs/locus-yield.md new file mode 100644 index 00000000..e225bc32 --- /dev/null +++ b/labs/22-yield/yield/docs/locus-yield.md @@ -0,0 +1,61 @@ +# Locus derivation — the Yield run lifecycle + +Formal grounding for the V1 architecture. Models and the candidate +derivation live in `docs/locus/`; they were fidelity-certified against the +design artifact before implementation, so every claim below is a theorem +about the design model, discharged into system claims by the tests named +here. + +## Models + +| model | operators | verdict | +|---|---|---| +| `yield-protocol.json` | control.supervisory-rw | **controllable** — violations: `[]` | +| `yield-protocol.json` | control.nonblockingness | **nonblocking** — blocking states: `[]` | +| `yield-diag-portable.json` | control.diagnosability | **not diagnosable** — witness below | +| `yield-diag-correlated.json` | control.diagnosability | **diagnosable** — no indistinguishable pairs | + +## What the theorems fixed in the design + +1. **Protocol integrity is supervisable.** With `accept_stale` and + `complete_unproven` modeled as controllable forbidden transitions, a + supervisor can always prevent stale-response acceptance and + completion-after-failed-requirement. The kernel's obligations name the + refusing mechanisms the implementation must own; they are discharged by + the refusing tests in `internal/guard/guard_test.go`: + - `disable-mechanism:accept_stale` → `TestRefusesStaleResponse`, + `TestRefusesDuplicateWithDifferentContent` + - `disable-mechanism:accept_response` → `TestRefusesSchemaInvalidResult` + - `disable-mechanism:complete_unproven` → + `TestRefusesCompletionAfterFailedRequirement`, + `engine.TestFailedRequirementBlocksRun` + - `disable-mechanism:complete` → `TestAllowsCompletionWithPassedRequirements` + +2. **Every run reaches an honest terminal — because the migrate verb + exists.** Nonblockingness holds only with two controllable exits from + `DIVERGED`: `migrate_digest` (`resume --accept-new-digest`) and + `declare_blocked`. Without the migrate verb the design blocks; it is + load-bearing, not a convenience. Discharged by + `engine.TestDigestMismatchRefusedThenMigrates` and + `engine.TestReplayDivergenceFailsLoudly`. + +3. **Portable mode is provably non-diagnosable for off-protocol agent + action.** Verbatim witness (indistinguishable pair): faulty + `OFF_PROTOCOL` vs normal `PENDING_OP` — a run where the agent acted + outside the yielded operation produces the same observable trace as an + honest one. This is why the README's "not guaranteed" column exists, + and why a correlated host adapter (same alphabet, `agent_off_protocol` + observable) is the principled post-V1 slice: the rival-design + derivation (`drv-998aec…`) shows it diagnosable with zero + indistinguishable pairs. + +## Claim scope + +The models were certified against the design description, not a running +system; the run lifecycle claims descend to the implementation exactly as +far as the named tests carry them. The remaining undischarged honesty gap +is recorded in each model's `unknowns`: a schema-valid `agent_task` result +can still be fabricated — schema validity is not truth. `run_command` is +the exception by construction: the engine executes commands itself, so +their results enter the log as observed fact +(`engine.TestEndToEndRunResumeComplete` asserts the observed output). diff --git a/labs/22-yield/yield/examples/investigate/SKILL.md b/labs/22-yield/yield/examples/investigate/SKILL.md new file mode 100644 index 00000000..993a4f5a --- /dev/null +++ b/labs/22-yield/yield/examples/investigate/SKILL.md @@ -0,0 +1,22 @@ +--- +name: investigate +description: Investigate a failure with bounded hypotheses, cheap-first probes, and an evidence-bound conclusion. +--- + +Run: + + yskill run . + +Follow each returned operation exactly. + +- `ask_user`: ask the user using the host's normal interface. +- `agent_task`: perform the task and return schema-valid JSON. +- `run_command`: yskill executes it itself; you will not see this kind. + +Resume the run after each operation: + + yskill resume --response response.json --skill . + +Do not skip an operation or invent its response. The program bounds the +investigation: at least three hypotheses, cheapest-to-disprove first, at +most three failed attempts, and completion requires a causal chain. diff --git a/labs/22-yield/yield/examples/investigate/fixtures/responses.json b/labs/22-yield/yield/examples/investigate/fixtures/responses.json new file mode 100644 index 00000000..3c8cb0a1 --- /dev/null +++ b/labs/22-yield/yield/examples/investigate/fixtures/responses.json @@ -0,0 +1,35 @@ +{ + "collect-evidence": { + "observations": [ + "CI fails on ubuntu only with 'Text file busy' (exit 126)", + "failure started after the hydrate step became concurrent", + "macOS and windows runners are green" + ] + }, + "form-hypotheses": { + "hypotheses": [ + { + "id": "h1", + "statement": "The runner image is missing the binary entirely", + "disprove_command": "exit 1" + }, + { + "id": "h2", + "statement": "Concurrent hydrate writes the binary while another process execs it (ETXTBSY)", + "disprove_command": "true" + }, + { + "id": "h3", + "statement": "A permissions regression strips the execute bit", + "disprove_command": "true" + } + ] + }, + "assess-h1": { + "refuted": true + }, + "assess-h2": { + "refuted": false, + "causal_chain": "concurrent hydrate holds the binary open for write -> exec of the same inode returns ETXTBSY -> shell reports exit 126 -> job fails only where hydrate and exec overlap (ubuntu)" + } +} diff --git a/labs/22-yield/yield/examples/investigate/main.go b/labs/22-yield/yield/examples/investigate/main.go new file mode 100644 index 00000000..dc51f950 --- /dev/null +++ b/labs/22-yield/yield/examples/investigate/main.go @@ -0,0 +1,99 @@ +// The reference skill: investigate a failure with bounded, evidence-bound +// discipline. The flow — the part prose skills lose under context pressure +// — is code: at least three hypotheses, cheapest-to-disprove first, at +// most three failed attempts, completion requires a causal chain. +// Judgment — forming and assessing hypotheses — stays with the model. +package main + +import ( + "encoding/json" + "fmt" + + "yield/sdk/yield" +) + +type hypothesis struct { + ID string `json:"id"` + Statement string `json:"statement"` + DisproveCommand string `json:"disprove_command"` +} + +type assessment struct { + Refuted bool `json:"refuted"` + CausalChain string `json:"causal_chain"` +} + +const hypothesesSchema = `{ + "type": "object", + "required": ["hypotheses"], + "properties": { + "hypotheses": { + "type": "array", + "minItems": 3, + "items": { + "type": "object", + "required": ["id", "statement", "disprove_command"], + "properties": { + "id": {"type": "string"}, + "statement": {"type": "string"}, + "disprove_command": {"type": "string"} + } + } + } + } +}` + +const assessmentSchema = `{ + "type": "object", + "required": ["refuted"], + "properties": { + "refuted": {"type": "boolean"}, + "causal_chain": {"type": "string"} + } +}` + +func main() { + yield.Main(func(ctx *yield.Context) (yield.Outcome, error) { + evidence := ctx.AgentTask("collect-evidence", + "Collect the observable evidence for the failure under investigation: error output, logs, recent changes. Return {\"observations\": [string]}.", + nil, json.RawMessage(`{"type":"object","required":["observations"],"properties":{"observations":{"type":"array","items":{"type":"string"}}}}`)) + + raw := ctx.AgentTask("form-hypotheses", + "Produce at least three hypotheses explaining the evidence, ordered cheapest-to-disprove first. Each carries a shell command whose failure would disprove it.", + json.RawMessage(evidence), json.RawMessage(hypothesesSchema)) + var hs struct { + Hypotheses []hypothesis `json:"hypotheses"` + } + if err := json.Unmarshal(raw, &hs); err != nil { + return yield.Outcome{}, err + } + + failures := 0 + for _, h := range hs.Hypotheses { + if failures >= 3 { + break + } + result := ctx.RunCommand("probe-"+h.ID, h.DisproveCommand, 300) + assessRaw := ctx.AgentTask("assess-"+h.ID, + fmt.Sprintf("Hypothesis %q: %s. Given the probe result, is it refuted? If it survives, state the causal chain from root cause to observed failure.", h.ID, h.Statement), + map[string]any{"hypothesis": h, "probe": result}, + json.RawMessage(assessmentSchema)) + var a assessment + if err := json.Unmarshal(assessRaw, &a); err != nil { + return yield.Outcome{}, err + } + if a.Refuted { + failures++ + continue + } + ctx.Require(a.CausalChain != "", "the surviving hypothesis states a causal chain", a) + return ctx.Complete(map[string]any{ + "hypothesis": h, + "causal_chain": a.CausalChain, + "probe_exit": result.ExitCode, + }) + } + return yield.Outcome{}, ctx.Blocked( + fmt.Sprintf("frontier reached: %d hypotheses refuted with %d failed attempts and none surviving — new evidence is needed, not more guessing", len(hs.Hypotheses), failures)) + }) +} diff --git a/labs/22-yield/yield/go.mod b/labs/22-yield/yield/go.mod new file mode 100644 index 00000000..a045a6fb --- /dev/null +++ b/labs/22-yield/yield/go.mod @@ -0,0 +1,7 @@ +module yield + +go 1.26.5 + +require github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 + +require golang.org/x/text v0.14.0 // indirect diff --git a/labs/22-yield/yield/go.sum b/labs/22-yield/yield/go.sum new file mode 100644 index 00000000..8fee20f8 --- /dev/null +++ b/labs/22-yield/yield/go.sum @@ -0,0 +1,6 @@ +github.com/dlclark/regexp2 v1.11.0 h1:G/nrcoOa7ZXlpoa/91N3X7mM3r8eIlMBBJZvsz/mxKI= +github.com/dlclark/regexp2 v1.11.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= +github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 h1:KRzFb2m7YtdldCEkzs6KqmJw4nqEVZGK7IN2kJkjTuQ= +github.com/santhosh-tekuri/jsonschema/v6 v6.0.2/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU= +golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= diff --git a/labs/22-yield/yield/internal/engine/engine.go b/labs/22-yield/yield/internal/engine/engine.go new file mode 100644 index 00000000..79641b61 --- /dev/null +++ b/labs/22-yield/yield/internal/engine/engine.go @@ -0,0 +1,397 @@ +// Package engine is the supervisor: it owns run creation, subprocess +// execution, the auto-execution of run_command operations, response +// acceptance, and terminal handling. Every state change goes through the +// run log; every refusal goes through the guard. +package engine + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "sort" + "strings" + "time" + + "yield/internal/guard" + "yield/internal/protocol" + "yield/internal/runlog" +) + +// Engine binds a skill directory to a runs directory. +type Engine struct { + SkillDir string + RunsDir string + // Stderr receives subprocess diagnostics (compile errors etc.). + Stderr *os.File +} + +// New creates an engine rooted at the skill directory; run logs live in +// /.yield/runs. +func New(skillDir string) (*Engine, error) { + abs, err := filepath.Abs(skillDir) + if err != nil { + return nil, err + } + runs, err := runlog.RunsDir(abs) + if err != nil { + return nil, err + } + return &Engine{SkillDir: abs, RunsDir: runs, Stderr: os.Stderr}, nil +} + +// Advance's result: either the next operation for the agent, or a +// terminal status. +type Progress struct { + RunID string + Envelope *protocol.RequestEnvelope + Terminal *protocol.TerminalOutcome +} + +// StartRun creates a run bound to the current skill digest and advances +// to the first agent-facing operation or terminal. +func (e *Engine) StartRun(input json.RawMessage) (*Progress, error) { + digest, err := protocol.DigestSkillDir(e.SkillDir) + if err != nil { + return nil, err + } + skill := protocol.SkillRef{Name: filepath.Base(e.SkillDir), Digest: digest} + runID := newRunID() + l, err := runlog.Create(e.RunsDir, runID) + if err != nil { + return nil, err + } + if _, err := l.Append(runlog.RunStarted, map[string]any{ + "run_id": runID, "skill": skill, + "input_digest": protocol.DigestBytes(input), + }); err != nil { + return nil, err + } + return e.advance(l, runID) +} + +// Resume validates and accepts a response for the pending operation, then +// advances. migrate=true explicitly rebinds the run to the current skill +// digest (the migrate_digest mechanism; divergence detection remains the +// safety net). +func (e *Engine) Resume(runID string, respBytes []byte, migrate bool) (*Progress, error) { + l, err := runlog.Open(e.RunsDir, runID) + if err != nil { + return nil, err + } + s, err := guard.Reconstruct(l) + if err != nil { + return nil, err + } + current, err := protocol.DigestSkillDir(e.SkillDir) + if err != nil { + return nil, err + } + if err := guard.CheckDigest(s, current, migrate); err != nil { + return nil, e.rejected(l, err) + } + if migrate && s.BoundDigest != current { + if _, err := l.Append(runlog.DigestMigrated, map[string]string{"from": s.BoundDigest, "to": current}); err != nil { + return nil, err + } + } + var resp protocol.ResponseEnvelope + if err := json.Unmarshal(respBytes, &resp); err != nil { + return nil, fmt.Errorf("response does not decode: %w", err) + } + if err := guard.CheckResponse(s, resp); err != nil { + return nil, e.rejected(l, err) + } + if err := e.acceptResponse(l, s.Pending, resp); err != nil { + return nil, err + } + return e.advance(l, runID) +} + +// Replay re-executes the program against the full journal and verifies it +// reproduces the run's recorded frontier — the determinism check. +func (e *Engine) Replay(runID string) (*Progress, error) { + l, err := runlog.Open(e.RunsDir, runID) + if err != nil { + return nil, err + } + s, err := guard.Reconstruct(l) + if err != nil { + return nil, err + } + out, err := e.execute(l, runID) + if err != nil { + return nil, err + } + switch out.Type { + case protocol.OutputDiverged: + return nil, fmt.Errorf("replay diverged at sequence %d: %s", out.Divergence.Sequence, out.Divergence.Detail) + case protocol.OutputRequest: + if s.Pending == nil { + return nil, fmt.Errorf("replay reached a new operation (seq %d) but the log has no pending operation", out.Envelope.Sequence) + } + if protocol.RequestDigest(out.Envelope.Request) != protocol.RequestDigest(s.Pending.Request) { + return nil, fmt.Errorf("replay reached a different pending operation than recorded") + } + return &Progress{RunID: runID, Envelope: out.Envelope}, nil + case protocol.OutputTerminal: + return &Progress{RunID: runID, Terminal: out.Terminal}, nil + } + return nil, fmt.Errorf("program emitted unknown output type %q", out.Type) +} + +// Log opens a run's log for inspection. +func (e *Engine) Log(runID string) (*runlog.Log, error) { + return runlog.Open(e.RunsDir, runID) +} + +// ListRuns returns known run IDs, newest last. +func (e *Engine) ListRuns() ([]string, error) { + entries, err := os.ReadDir(e.RunsDir) + if err != nil { + return nil, err + } + var ids []string + for _, en := range entries { + if strings.HasSuffix(en.Name(), ".jsonl") { + ids = append(ids, strings.TrimSuffix(en.Name(), ".jsonl")) + } + } + sort.Strings(ids) + return ids, nil +} + +// advance executes the program and drives it as far as it can go without +// the agent: run_command operations are executed by the engine itself +// (observed fact), everything else is handed back as the next envelope. +func (e *Engine) advance(l *runlog.Log, runID string) (*Progress, error) { + for { + out, err := e.execute(l, runID) + if err != nil { + return nil, err + } + switch out.Type { + case protocol.OutputDiverged: + if _, err := l.Append(runlog.ReplayDiverged, out.Divergence); err != nil { + return nil, err + } + return nil, fmt.Errorf("replay diverged at sequence %d: %s — the skill program changed behavior mid-run; resume with --accept-new-digest only rebinds sources, divergence always fails the run", out.Divergence.Sequence, out.Divergence.Detail) + + case protocol.OutputRequest: + env := out.Envelope + s, err := guard.Reconstruct(l) + if err != nil { + return nil, err + } + // Idempotent re-emission of the already-pending operation. + if s.Pending == nil || protocol.RequestDigest(s.Pending.Request) != protocol.RequestDigest(env.Request) || s.Pending.Sequence != env.Sequence { + if _, err := l.Append(runlog.OperationRequested, env); err != nil { + return nil, err + } + } + if env.Request.Kind == protocol.OpRunCommand { + resp, err := e.runCommand(env) + if err != nil { + return nil, err + } + if err := e.acceptResponse(l, env, resp); err != nil { + return nil, err + } + continue // the program can now advance past this operation + } + return &Progress{RunID: runID, Envelope: env}, nil + + case protocol.OutputTerminal: + return e.terminate(l, runID, out) + + default: + return nil, fmt.Errorf("program emitted unknown output type %q", out.Type) + } + } +} + +// execute runs the skill subprocess once against the journal rebuilt from +// the log and returns its single ProgramOutput. +func (e *Engine) execute(l *runlog.Log, runID string) (*protocol.ProgramOutput, error) { + s, err := guard.Reconstruct(l) + if err != nil { + return nil, err + } + journal := protocol.Journal{RunID: runID, Skill: s.Skill} + // Rebuild answered entries in sequence order from the log. + pendingBySeq := map[int]protocol.Request{} + results := map[int]json.RawMessage{} + for _, ev := range l.Events() { + switch ev.Type { + case runlog.OperationRequested: + var env protocol.RequestEnvelope + if err := ev.Decode(&env); err != nil { + return nil, err + } + pendingBySeq[env.Sequence] = env.Request + case runlog.OperationCompleted: + var d struct { + Sequence int `json:"sequence"` + Result json.RawMessage `json:"result"` + } + if err := ev.Decode(&d); err != nil { + return nil, err + } + results[d.Sequence] = d.Result + } + } + for seq := 1; ; seq++ { + req, ok := pendingBySeq[seq] + res, done := results[seq] + if !ok || !done { + break + } + journal.Entries = append(journal.Entries, protocol.JournalEntry{ + Request: req, + Response: protocol.ResponseEnvelope{ + RunID: runID, Sequence: seq, RequestID: req.ID, Status: "completed", Result: res, + }, + }) + } + jf, err := os.CreateTemp("", "yield-journal-*.json") + if err != nil { + return nil, err + } + defer os.Remove(jf.Name()) + if err := json.NewEncoder(jf).Encode(journal); err != nil { + return nil, err + } + jf.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + cmd := exec.CommandContext(ctx, "go", "run", ".") + cmd.Dir = e.SkillDir + cmd.Env = append(os.Environ(), "YIELD_JOURNAL="+jf.Name()) + cmd.Stderr = e.Stderr + outBytes, err := cmd.Output() + if err != nil { + return nil, fmt.Errorf("skill program failed: %w", err) + } + var out protocol.ProgramOutput + if err := json.Unmarshal(outBytes, &out); err != nil { + return nil, fmt.Errorf("skill program emitted invalid output: %w", err) + } + return &out, nil +} + +// runCommand executes a run_command operation itself with a timeout; the +// result enters the log as observed fact. +func (e *Engine) runCommand(env *protocol.RequestEnvelope) (protocol.ResponseEnvelope, error) { + var p protocol.RunCommandPayload + if err := json.Unmarshal(env.Request.Payload, &p); err != nil { + return protocol.ResponseEnvelope{}, err + } + timeout := time.Duration(p.TimeoutSeconds) * time.Second + if timeout <= 0 { + timeout = 10 * time.Minute + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + cmd := exec.CommandContext(ctx, "sh", "-c", p.Command) + cmd.Dir = e.SkillDir + var stdout, stderr strings.Builder + cmd.Stdout = &stdout + cmd.Stderr = &stderr + runErr := cmd.Run() + res := protocol.CommandResult{Stdout: stdout.String(), Stderr: stderr.String()} + if ctx.Err() == context.DeadlineExceeded { + res.TimedOut = true + res.ExitCode = -1 + } else if exitErr, ok := runErr.(*exec.ExitError); ok { + res.ExitCode = exitErr.ExitCode() + } else if runErr != nil { + return protocol.ResponseEnvelope{}, runErr + } + raw, err := json.Marshal(res) + if err != nil { + return protocol.ResponseEnvelope{}, err + } + return protocol.ResponseEnvelope{ + RunID: env.RunID, Sequence: env.Sequence, RequestID: env.Request.ID, + Status: "completed", Result: raw, + }, nil +} + +// acceptResponse appends an accepted response as operation.completed. +func (e *Engine) acceptResponse(l *runlog.Log, env *protocol.RequestEnvelope, resp protocol.ResponseEnvelope) error { + _, err := l.Append(runlog.OperationCompleted, map[string]any{ + "sequence": resp.Sequence, + "request_id": resp.RequestID, + "result": resp.Result, + "result_digest": protocol.DigestBytes(resp.Result), + }) + return err +} + +// terminate closes the run, enforcing evidence-bound completion. +func (e *Engine) terminate(l *runlog.Log, runID string, out *protocol.ProgramOutput) (*Progress, error) { + s, err := guard.Reconstruct(l) + if err != nil { + return nil, err + } + term := out.Terminal + // Record the requirement trail first. + for _, r := range out.Requirements { + t := runlog.RequirementPassed + if !r.Passed { + t = runlog.RequirementFailed + } + if _, err := l.Append(t, r); err != nil { + return nil, err + } + } + switch term.Status { + case protocol.StatusCompleted: + if err := guard.CheckCompletion(s, out.Requirements); err != nil { + // complete_unproven is forbidden: the run closes blocked, loudly. + if _, aerr := l.Append(runlog.RunBlocked, map[string]string{"reason": err.Error()}); aerr != nil { + return nil, aerr + } + return nil, err + } + if _, err := l.Append(runlog.RunCompleted, map[string]any{ + "result": term.Result, "requirements": len(out.Requirements), + }); err != nil { + return nil, err + } + case protocol.StatusRequirementFailed, protocol.StatusBlocked: + if _, err := l.Append(runlog.RunBlocked, map[string]string{"reason": term.Reason}); err != nil { + return nil, err + } + case protocol.StatusRefused: + if _, err := l.Append(runlog.RunRefused, map[string]string{"reason": term.Reason}); err != nil { + return nil, err + } + default: + return nil, fmt.Errorf("program emitted unknown terminal status %q", term.Status) + } + return &Progress{RunID: runID, Terminal: term}, nil +} + +// rejected records a guard refusal in the log and returns it. +func (e *Engine) rejected(l *runlog.Log, err error) error { + if rej, ok := err.(*guard.Rejection); ok { + _, _ = l.Append(runlog.ResponseRejected, map[string]string{ + "reason": string(rej.Reason), "detail": rej.Detail, + }) + } + return err +} + +func newRunID() string { + var b [5]byte + if _, err := rand.Read(b[:]); err != nil { + panic(err) + } + return fmt.Sprintf("run_%d_%s", time.Now().UTC().Unix(), hex.EncodeToString(b[:])) +} diff --git a/labs/22-yield/yield/internal/engine/engine_test.go b/labs/22-yield/yield/internal/engine/engine_test.go new file mode 100644 index 00000000..8bd09791 --- /dev/null +++ b/labs/22-yield/yield/internal/engine/engine_test.go @@ -0,0 +1,277 @@ +package engine + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + + "yield/internal/guard" + "yield/internal/protocol" + "yield/internal/runlog" +) + +// testEngine points at a testdata skill but keeps run logs in a temp dir, +// so tests never write into the source tree. +func testEngine(t *testing.T, skill string) *Engine { + t.Helper() + abs, err := filepath.Abs(filepath.Join("testdata", skill)) + if err != nil { + t.Fatal(err) + } + return &Engine{SkillDir: abs, RunsDir: t.TempDir(), Stderr: os.Stderr} +} + +func respond(t *testing.T, e *Engine, p *Progress, result string, migrate bool) (*Progress, error) { + t.Helper() + b, err := json.Marshal(protocol.ResponseEnvelope{ + RunID: p.RunID, Sequence: p.Envelope.Sequence, + RequestID: p.Envelope.Request.ID, Status: "completed", + Result: json.RawMessage(result), + }) + if err != nil { + t.Fatal(err) + } + return e.Resume(p.RunID, b, migrate) +} + +func TestEndToEndRunResumeComplete(t *testing.T) { + e := testEngine(t, "skill-basic") + + p, err := e.StartRun(nil) + if err != nil { + t.Fatal(err) + } + if p.Envelope == nil || p.Envelope.Request.ID != "confirm-scope" || p.Envelope.Request.Kind != protocol.OpAskUser { + t.Fatalf("first operation must be ask_user confirm-scope, got %+v", p.Envelope) + } + if p.Envelope.Protocol != protocol.Version { + t.Fatalf("envelope must carry %s", protocol.Version) + } + + // A stale response (sequence 0) is refused and recorded. + stale, _ := json.Marshal(protocol.ResponseEnvelope{ + RunID: p.RunID, Sequence: 0, RequestID: "confirm-scope", + Status: "completed", Result: json.RawMessage(`{"value":"x"}`), + }) + if _, err := e.Resume(p.RunID, stale, false); err == nil { + t.Fatal("stale response must be refused") + } + + p, err = respond(t, e, p, `{"value":"preserve"}`, false) + if err != nil { + t.Fatal(err) + } + if p.Envelope == nil || p.Envelope.Request.ID != "summarize" { + t.Fatalf("second operation must be agent_task summarize, got %+v", p.Envelope) + } + + // A schema-invalid agent_task result is refused. + if _, err := respond(t, e, p, `{"not_summary":1}`, false); err == nil { + t.Fatal("schema-invalid result must be refused") + } + + // A valid result lets the engine advance through run_command (executed + // by the engine itself) and the requirement to completion. + p, err = respond(t, e, p, `{"summary":"a tiny repo"}`, false) + if err != nil { + t.Fatal(err) + } + if p.Terminal == nil || p.Terminal.Status != protocol.StatusCompleted { + t.Fatalf("run must complete, got %+v", p) + } + + l, err := e.Log(p.RunID) + if err != nil { + t.Fatal(err) + } + var types []string + var cmdResult protocol.CommandResult + for _, ev := range l.Events() { + types = append(types, string(ev.Type)) + if ev.Type == runlog.OperationCompleted { + var d struct { + RequestID string `json:"request_id"` + Result json.RawMessage `json:"result"` + } + _ = ev.Decode(&d) + if d.RequestID == "run-tests" { + _ = json.Unmarshal(d.Result, &cmdResult) + } + } + } + joined := strings.Join(types, ",") + for _, want := range []string{"run.started", "operation.requested", "operation.completed", "requirement.passed", "run.completed"} { + if !strings.Contains(joined, want) { + t.Fatalf("log must contain %s; got %s", want, joined) + } + } + if strings.Contains(joined, string(runlog.RequirementFailed)) { + t.Fatalf("no requirement failed in this run; got %s", joined) + } + // run_command was executed by the engine: observed fact, not transcription. + if cmdResult.ExitCode != 0 || !strings.Contains(cmdResult.Stdout, "test-ok") { + t.Fatalf("run-tests result must be the observed command output, got %+v", cmdResult) + } + + // The closed run refuses further responses. + if _, err := respond(t, e, &Progress{RunID: p.RunID, Envelope: &protocol.RequestEnvelope{Sequence: 2, Request: protocol.Request{ID: "summarize"}}}, `{"summary":"again"}`, false); err == nil { + t.Fatal("responses on a closed run must be refused") + } +} + +func TestReplayIsDeterministic(t *testing.T) { + e := testEngine(t, "skill-basic") + p, err := e.StartRun(nil) + if err != nil { + t.Fatal(err) + } + p, err = respond(t, e, p, `{"value":"migration"}`, false) + if err != nil { + t.Fatal(err) + } + rp, err := e.Replay(p.RunID) + if err != nil { + t.Fatalf("replay must be deterministic: %v", err) + } + if rp.Envelope == nil || rp.Envelope.Request.ID != p.Envelope.Request.ID { + t.Fatalf("replay must reach the recorded frontier %q, got %+v", p.Envelope.Request.ID, rp) + } +} + +func TestReplayDivergenceFailsLoudly(t *testing.T) { + e := testEngine(t, "skill-envbranch") + t.Setenv("YIELD_TEST_BRANCH", "a") + + p, err := e.StartRun(nil) + if err != nil { + t.Fatal(err) + } + p, err = respond(t, e, p, `{"value":"one"}`, false) + if err != nil { + t.Fatal(err) + } + if p.Envelope.Request.ID != "second-question-a" { + t.Fatalf("branch a must yield second-question-a, got %s", p.Envelope.Request.ID) + } + + // The program's behavior changes under its feet: replaying the journal + // now produces a different second operation. The run must fail loudly, + // never silently fork. + t.Setenv("YIELD_TEST_BRANCH", "b") + _, err = respond(t, e, p, `{"value":"two"}`, false) + if err == nil || !strings.Contains(err.Error(), "diverged") { + t.Fatalf("divergence must fail loudly, got %v", err) + } + + l, err := e.Log(p.RunID) + if err != nil { + t.Fatal(err) + } + found := false + for _, ev := range l.Events() { + if ev.Type == runlog.ReplayDiverged { + found = true + } + } + if !found { + t.Fatal("replay.diverged must be recorded in the run log") + } +} + +func TestFailedRequirementBlocksRun(t *testing.T) { + e := testEngine(t, "skill-reqfail") + p, err := e.StartRun(nil) + if err != nil { + t.Fatal(err) + } + if p.Terminal == nil || p.Terminal.Status == protocol.StatusCompleted { + t.Fatalf("a failed requirement must prevent completion, got %+v", p) + } + l, err := e.Log(p.RunID) + if err != nil { + t.Fatal(err) + } + s, err := guard.Reconstruct(l) + if err != nil { + t.Fatal(err) + } + if !s.ReqFailed || !s.Closed { + t.Fatalf("log must show requirement.failed and a closed run; state %+v", s) + } + var blocked bool + for _, ev := range l.Events() { + if ev.Type == runlog.RunBlocked { + blocked = true + } + if ev.Type == runlog.RunCompleted { + t.Fatal("run.completed must never follow a failed requirement") + } + } + if !blocked { + t.Fatal("run must close blocked") + } +} + +func TestDigestMismatchRefusedThenMigrates(t *testing.T) { + // Copy the basic skill into a mutable dir INSIDE the module tree (so + // `go run .` still resolves the yield module) and edit it mid-run; + // run logs stay in a separate temp dir. + src, err := filepath.Abs(filepath.Join("testdata", "skill-basic")) + if err != nil { + t.Fatal(err) + } + dir, err := os.MkdirTemp(filepath.Join("testdata"), "tmp-skill-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + if dir, err = filepath.Abs(dir); err != nil { + t.Fatal(err) + } + b, err := os.ReadFile(filepath.Join(src, "main.go")) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "main.go"), b, 0o644); err != nil { + t.Fatal(err) + } + e := &Engine{SkillDir: dir, RunsDir: t.TempDir(), Stderr: os.Stderr} + + p, err := e.StartRun(nil) + if err != nil { + t.Fatal(err) + } + + // Change the skill source mid-run: resume must refuse without migration. + if err := os.WriteFile(filepath.Join(dir, "main.go"), append(b, []byte("\n// edited mid-run\n")...), 0o644); err != nil { + t.Fatal(err) + } + _, err = respond(t, e, p, `{"value":"preserve"}`, false) + if err == nil || !strings.Contains(err.Error(), "digest-mismatch") { + t.Fatalf("digest mismatch must be refused, got %v", err) + } + // Explicit migration rebinds and proceeds (the edit is a comment, so + // replay does not diverge). + p2, err := respond(t, e, p, `{"value":"preserve"}`, true) + if err != nil { + t.Fatalf("explicit migration must proceed: %v", err) + } + if p2.Envelope == nil || p2.Envelope.Request.ID != "summarize" { + t.Fatalf("migrated run must advance, got %+v", p2) + } + l, err := e.Log(p.RunID) + if err != nil { + t.Fatal(err) + } + var migrated bool + for _, ev := range l.Events() { + if ev.Type == runlog.DigestMigrated { + migrated = true + } + } + if !migrated { + t.Fatal("digest.migrated must be recorded") + } +} diff --git a/labs/22-yield/yield/internal/engine/testdata/skill-basic/main.go b/labs/22-yield/yield/internal/engine/testdata/skill-basic/main.go new file mode 100644 index 00000000..9e9dfc64 --- /dev/null +++ b/labs/22-yield/yield/internal/engine/testdata/skill-basic/main.go @@ -0,0 +1,26 @@ +// Test skill: one of each primitive, evidence-bound completion. +package main + +import ( + "encoding/json" + + "yield/sdk/yield" +) + +func main() { + yield.Main(func(ctx *yield.Context) (yield.Outcome, error) { + scope := ctx.AskUser("confirm-scope", "May this change modify the public API?") + + summary := ctx.AgentTask("summarize", "Summarize the repository.", map[string]string{"scope": scope}, + json.RawMessage(`{"type":"object","required":["summary"],"properties":{"summary":{"type":"string"}}}`)) + + tests := ctx.RunCommand("run-tests", "echo test-ok", 30) + ctx.Require(tests.ExitCode == 0, "the test command passes", tests) + + var s struct { + Summary string `json:"summary"` + } + _ = json.Unmarshal(summary, &s) + return ctx.Complete(map[string]string{"scope": scope, "summary": s.Summary}) + }) +} diff --git a/labs/22-yield/yield/internal/engine/testdata/skill-envbranch/main.go b/labs/22-yield/yield/internal/engine/testdata/skill-envbranch/main.go new file mode 100644 index 00000000..51aaf689 --- /dev/null +++ b/labs/22-yield/yield/internal/engine/testdata/skill-envbranch/main.go @@ -0,0 +1,21 @@ +// Test skill that is DELIBERATELY nondeterministic across executions when +// YIELD_TEST_BRANCH changes: used to prove replay divergence fails loudly. +package main + +import ( + "os" + + "yield/sdk/yield" +) + +func main() { + yield.Main(func(ctx *yield.Context) (yield.Outcome, error) { + first := ctx.AskUser("first-question", "First?") + opID := "second-question-a" + if os.Getenv("YIELD_TEST_BRANCH") == "b" { + opID = "second-question-b" + } + second := ctx.AskUser(opID, "Second?") + return ctx.Complete(map[string]string{"first": first, "second": second}) + }) +} diff --git a/labs/22-yield/yield/internal/engine/testdata/skill-reqfail/main.go b/labs/22-yield/yield/internal/engine/testdata/skill-reqfail/main.go new file mode 100644 index 00000000..6d0651d8 --- /dev/null +++ b/labs/22-yield/yield/internal/engine/testdata/skill-reqfail/main.go @@ -0,0 +1,15 @@ +// Test skill whose requirement fails: the run must close blocked, never +// completed. +package main + +import ( + "yield/sdk/yield" +) + +func main() { + yield.Main(func(ctx *yield.Context) (yield.Outcome, error) { + tests := ctx.RunCommand("run-tests", "exit 3", 30) + ctx.Require(tests.ExitCode == 0, "the test command passes", tests) + return ctx.Complete(map[string]string{"unreachable": "yes"}) + }) +} diff --git a/labs/22-yield/yield/internal/guard/guard.go b/labs/22-yield/yield/internal/guard/guard.go new file mode 100644 index 00000000..04094a21 --- /dev/null +++ b/labs/22-yield/yield/internal/guard/guard.go @@ -0,0 +1,176 @@ +// Package guard owns every refusal in the protocol. Each rejection reason +// corresponds to a supervisory obligation from the Locus derivation +// (docs/locus): the controllability theorem rests on these events being +// genuinely refusable, so each has a named check and a refusing test. +package guard + +import ( + "fmt" + + "yield/internal/protocol" + "yield/internal/runlog" +) + +type RejectReason string + +const ( + RejectWrongRun RejectReason = "wrong-run" + RejectStale RejectReason = "stale-response" + RejectDuplicate RejectReason = "duplicate-response" + RejectWrongRequest RejectReason = "wrong-request" + RejectSchemaInvalid RejectReason = "schema-invalid" + RejectDigestMismatch RejectReason = "digest-mismatch" + RejectUnproven RejectReason = "completion-unproven" + RejectRunClosed RejectReason = "run-closed" + RejectNoPendingOp RejectReason = "no-pending-operation" +) + +// Rejection is a typed refusal; it renders calmly and names its reason. +type Rejection struct { + Reason RejectReason + Detail string +} + +func (r *Rejection) Error() string { + return fmt.Sprintf("rejected (%s): %s", r.Reason, r.Detail) +} + +func reject(reason RejectReason, format string, args ...any) *Rejection { + return &Rejection{Reason: reason, Detail: fmt.Sprintf(format, args...)} +} + +// RunState is the guard-relevant projection of a run log. +type RunState struct { + RunID string + BoundDigest string + Skill protocol.SkillRef + Pending *protocol.RequestEnvelope // unanswered operation, if any + Completed map[int]string // sequence -> result digest + Closed bool // a terminal run.* event exists + ReqFailed bool // a requirement.failed event exists + Diverged bool +} + +// Reconstruct folds a run log into its guard state. The log is the only +// source of truth; nothing else is consulted. +func Reconstruct(l *runlog.Log) (*RunState, error) { + s := &RunState{Completed: map[int]string{}} + for _, e := range l.Events() { + switch e.Type { + case runlog.RunStarted: + var d struct { + RunID string `json:"run_id"` + Skill protocol.SkillRef `json:"skill"` + } + if err := e.Decode(&d); err != nil { + return nil, err + } + s.RunID = d.RunID + s.Skill = d.Skill + s.BoundDigest = d.Skill.Digest + case runlog.OperationRequested: + var env protocol.RequestEnvelope + if err := e.Decode(&env); err != nil { + return nil, err + } + s.Pending = &env + case runlog.OperationCompleted: + var d struct { + Sequence int `json:"sequence"` + ResultDigest string `json:"result_digest"` + } + if err := e.Decode(&d); err != nil { + return nil, err + } + s.Completed[d.Sequence] = d.ResultDigest + if s.Pending != nil && s.Pending.Sequence == d.Sequence { + s.Pending = nil + } + case runlog.DigestMigrated: + var d struct { + To string `json:"to"` + } + if err := e.Decode(&d); err != nil { + return nil, err + } + s.BoundDigest = d.To + case runlog.RequirementFailed: + s.ReqFailed = true + case runlog.ReplayDiverged: + s.Diverged = true + case runlog.RunCompleted, runlog.RunBlocked, runlog.RunRefused: + s.Closed = true + } + } + return s, nil +} + +// CheckDigest refuses to act on a run whose skill source changed since the +// run was bound, unless the caller explicitly migrates. This is the +// migrate_digest mechanism from the lifecycle model — without it, a +// diverged-source run would block. +func CheckDigest(s *RunState, currentDigest string, migrate bool) error { + if s.BoundDigest == currentDigest { + return nil + } + if migrate { + return nil + } + return reject(RejectDigestMismatch, + "skill source changed since the run was bound (bound %s, current %s); re-run with --accept-new-digest to migrate explicitly", + short(s.BoundDigest), short(currentDigest)) +} + +// CheckResponse decides whether a response envelope may be accepted for +// the run's pending operation. Refusals: wrong run, stale, duplicate, +// wrong request id, schema-invalid result. +func CheckResponse(s *RunState, resp protocol.ResponseEnvelope) error { + if s.Closed { + return reject(RejectRunClosed, "run %s already reached a terminal state", s.RunID) + } + if resp.RunID != s.RunID { + return reject(RejectWrongRun, "response is for run %s, this run is %s", resp.RunID, s.RunID) + } + if s.Pending == nil { + return reject(RejectNoPendingOp, "run %s has no pending operation", s.RunID) + } + if resp.Sequence != s.Pending.Sequence { + if d, ok := s.Completed[resp.Sequence]; ok { + if d == protocol.DigestBytes(resp.Result) { + return reject(RejectDuplicate, "sequence %d already completed with identical content", resp.Sequence) + } + return reject(RejectDuplicate, "sequence %d already completed with DIFFERENT content; refusing to rewrite history", resp.Sequence) + } + return reject(RejectStale, "response targets sequence %d but the pending operation is sequence %d", resp.Sequence, s.Pending.Sequence) + } + if resp.RequestID != s.Pending.Request.ID { + return reject(RejectWrongRequest, "response is for request %q but the pending request is %q", resp.RequestID, s.Pending.Request.ID) + } + if resp.Status == "completed" { + if err := protocol.ValidateResult(s.Pending.Request.OutputSchema, resp.Result); err != nil { + return reject(RejectSchemaInvalid, "%v", err) + } + } + return nil +} + +// CheckCompletion refuses run.completed when any requirement has failed: +// evidence-bound completion is a supervisory invariant, not advice. +func CheckCompletion(s *RunState, reqs []protocol.Requirement) error { + if s.ReqFailed { + return reject(RejectUnproven, "a requirement already failed in this run; completion is refused") + } + for _, r := range reqs { + if !r.Passed { + return reject(RejectUnproven, "requirement %q failed; completion is refused", r.Claim) + } + } + return nil +} + +func short(digest string) string { + if len(digest) > 19 { + return digest[:19] + "…" + } + return digest +} diff --git a/labs/22-yield/yield/internal/guard/guard_test.go b/labs/22-yield/yield/internal/guard/guard_test.go new file mode 100644 index 00000000..d8050069 --- /dev/null +++ b/labs/22-yield/yield/internal/guard/guard_test.go @@ -0,0 +1,140 @@ +package guard + +import ( + "encoding/json" + "strings" + "testing" + + "yield/internal/protocol" +) + +// These tests are the discharge of the supervisory obligations from the +// Locus derivation (docs/locus): each controllable event the theorem +// rests on has a genuine refusing mechanism, exhibited here. + +func pendingState() *RunState { + return &RunState{ + RunID: "run_1", + BoundDigest: "sha256:aaa", + Pending: &protocol.RequestEnvelope{ + Protocol: protocol.Version, + RunID: "run_1", + Sequence: 3, + Request: protocol.Request{ + ID: "confirm-scope", + Kind: protocol.OpAskUser, + OutputSchema: json.RawMessage( + `{"type":"object","required":["value"],"properties":{"value":{"type":"string"}}}`), + }, + }, + Completed: map[int]string{ + 1: protocol.DigestBytes([]byte(`{"value":"a"}`)), + 2: protocol.DigestBytes([]byte(`{"value":"b"}`)), + }, + } +} + +func resp(seq int, id, result string) protocol.ResponseEnvelope { + return protocol.ResponseEnvelope{ + RunID: "run_1", Sequence: seq, RequestID: id, + Status: "completed", Result: json.RawMessage(result), + } +} + +func wantReject(t *testing.T, err error, reason RejectReason) { + t.Helper() + rej, ok := err.(*Rejection) + if !ok { + t.Fatalf("want *Rejection(%s), got %v", reason, err) + } + if rej.Reason != reason { + t.Fatalf("want reason %s, got %s (%s)", reason, rej.Reason, rej.Detail) + } +} + +func TestRefusesStaleResponse(t *testing.T) { + // disable-mechanism:accept_stale — a stale response never enters replay. + err := CheckResponse(pendingState(), resp(0, "confirm-scope", `{"value":"x"}`)) + wantReject(t, err, RejectStale) +} + +func TestRefusesDuplicateWithDifferentContent(t *testing.T) { + err := CheckResponse(pendingState(), resp(2, "confirm-scope", `{"value":"REWRITTEN"}`)) + wantReject(t, err, RejectDuplicate) + if !strings.Contains(err.Error(), "DIFFERENT content") { + t.Fatalf("rejection must name the history rewrite: %v", err) + } +} + +func TestRefusesDuplicateIdentical(t *testing.T) { + err := CheckResponse(pendingState(), resp(2, "confirm-scope", `{"value":"b"}`)) + wantReject(t, err, RejectDuplicate) +} + +func TestRefusesWrongRun(t *testing.T) { + r := resp(3, "confirm-scope", `{"value":"x"}`) + r.RunID = "run_OTHER" + wantReject(t, CheckResponse(pendingState(), r), RejectWrongRun) +} + +func TestRefusesWrongRequestID(t *testing.T) { + wantReject(t, CheckResponse(pendingState(), resp(3, "some-other-request", `{"value":"x"}`)), RejectWrongRequest) +} + +func TestRefusesSchemaInvalidResult(t *testing.T) { + // disable-mechanism:accept_response — schema validity gates acceptance. + wantReject(t, CheckResponse(pendingState(), resp(3, "confirm-scope", `{"wrong":"shape"}`)), RejectSchemaInvalid) +} + +func TestAcceptsValidResponse(t *testing.T) { + if err := CheckResponse(pendingState(), resp(3, "confirm-scope", `{"value":"preserve"}`)); err != nil { + t.Fatalf("valid response must be accepted: %v", err) + } +} + +func TestRefusesResponseOnClosedRun(t *testing.T) { + s := pendingState() + s.Closed = true + wantReject(t, CheckResponse(s, resp(3, "confirm-scope", `{"value":"x"}`)), RejectRunClosed) +} + +func TestRefusesCompletionAfterFailedRequirement(t *testing.T) { + // disable-mechanism:complete_unproven — the forbidden transition + // REQ_FAILED --complete--> COMPLETED is structurally refused. + s := pendingState() + s.ReqFailed = true + wantReject(t, CheckCompletion(s, nil), RejectUnproven) + + err := CheckCompletion(pendingState(), []protocol.Requirement{ + {Claim: "tests pass", Passed: false}, + }) + wantReject(t, err, RejectUnproven) +} + +func TestAllowsCompletionWithPassedRequirements(t *testing.T) { + // disable-mechanism:complete — completion is permitted exactly when + // every requirement passed. + err := CheckCompletion(pendingState(), []protocol.Requirement{ + {Claim: "tests pass", Passed: true}, + }) + if err != nil { + t.Fatalf("completion with passed requirements must be allowed: %v", err) + } +} + +func TestRefusesDigestMismatchWithoutMigration(t *testing.T) { + s := pendingState() + if err := CheckDigest(s, "sha256:aaa", false); err != nil { + t.Fatalf("matching digest must pass: %v", err) + } + err := CheckDigest(s, "sha256:bbb", false) + wantReject(t, err, RejectDigestMismatch) + if !strings.Contains(err.Error(), "--accept-new-digest") { + t.Fatalf("rejection must prescribe the migrate verb: %v", err) + } + // migrate_digest is the controllable recovery that keeps DIVERGED + // non-blocking in the lifecycle model. + if err := CheckDigest(s, "sha256:bbb", true); err != nil { + t.Fatalf("explicit migration must be allowed: %v", err) + } +} diff --git a/labs/22-yield/yield/internal/protocol/protocol.go b/labs/22-yield/yield/internal/protocol/protocol.go new file mode 100644 index 00000000..12f7fc77 --- /dev/null +++ b/labs/22-yield/yield/internal/protocol/protocol.go @@ -0,0 +1,272 @@ +// Package protocol defines the yield.v1 wire protocol: the typed operation +// envelopes a skill program yields to the coding agent, and the response +// envelopes the agent (or yskill itself) feeds back. Everything that crosses +// a process boundary is defined here and nowhere else. +package protocol + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io/fs" + "os" + "path/filepath" + "sort" + "strings" + + "github.com/santhosh-tekuri/jsonschema/v6" +) + +// Version is the protocol identifier carried by every request envelope. +const Version = "yield.v1" + +// OpKind is the closed set of operations a skill program may yield. +type OpKind string + +const ( + OpAskUser OpKind = "ask_user" + OpAgentTask OpKind = "agent_task" + OpRunCommand OpKind = "run_command" +) + +// SkillRef identifies the skill a run is bound to. Digest is the +// content digest of the skill source; responses against a different +// digest are rejected unless explicitly migrated. +type SkillRef struct { + Name string `json:"name"` + Version string `json:"version,omitempty"` + Digest string `json:"digest"` +} + +// Request is one yielded operation. +type Request struct { + ID string `json:"id"` + Kind OpKind `json:"kind"` + Payload json.RawMessage `json:"payload"` + OutputSchema json.RawMessage `json:"output_schema,omitempty"` +} + +// RequestEnvelope is what the agent sees: one operation, bound to a run, +// a sequence number, and the skill digest. +type RequestEnvelope struct { + Protocol string `json:"protocol"` + RunID string `json:"run_id"` + Skill SkillRef `json:"skill"` + Sequence int `json:"sequence"` + Request Request `json:"request"` +} + +// ResponseEnvelope is what comes back for exactly one pending request. +type ResponseEnvelope struct { + RunID string `json:"run_id"` + Sequence int `json:"sequence"` + RequestID string `json:"request_id"` + Status string `json:"status"` // "completed" | "failed" + Result json.RawMessage `json:"result"` +} + +// AskUserPayload asks a question through the host's normal interface. +type AskUserPayload struct { + Question string `json:"question"` + Options []Option `json:"options,omitempty"` +} + +type Option struct { + Value string `json:"value"` + Label string `json:"label,omitempty"` +} + +// AskUserResult is the expected result shape for ask_user. +type AskUserResult struct { + Value string `json:"value"` +} + +// AgentTaskPayload delegates a reasoning task to the model. +type AgentTaskPayload struct { + Instruction string `json:"instruction"` + Context json.RawMessage `json:"context,omitempty"` +} + +// RunCommandPayload names a command yskill executes itself, so the result +// enters the log as observed fact rather than the agent's transcription. +type RunCommandPayload struct { + Command string `json:"command"` + TimeoutSeconds int `json:"timeout_seconds,omitempty"` +} + +// CommandResult is the observed outcome of a run_command operation. +type CommandResult struct { + ExitCode int `json:"exit_code"` + Stdout string `json:"stdout"` + Stderr string `json:"stderr"` + TimedOut bool `json:"timed_out,omitempty"` +} + +// Requirement is a claim bound to evidence; a failed requirement prevents +// run completion. +type Requirement struct { + Claim string `json:"claim"` + Passed bool `json:"passed"` + EvidenceDigest string `json:"evidence_digest,omitempty"` +} + +// Program output types: a skill subprocess emits exactly one ProgramOutput +// per execution — the next yielded request, a terminal outcome, or a +// replay divergence report. +const ( + OutputRequest = "request" + OutputTerminal = "terminal" + OutputDiverged = "diverged" +) + +// Terminal statuses. +const ( + StatusCompleted = "completed" + StatusBlocked = "blocked" + StatusRefused = "refused" + StatusRequirementFailed = "requirement_failed" +) + +type ProgramOutput struct { + Type string `json:"type"` + Envelope *RequestEnvelope `json:"envelope,omitempty"` + Terminal *TerminalOutcome `json:"terminal,omitempty"` + Divergence *Divergence `json:"divergence,omitempty"` + Requirements []Requirement `json:"requirements,omitempty"` +} + +type TerminalOutcome struct { + Status string `json:"status"` + Result json.RawMessage `json:"result,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// Divergence reports that replay produced a different operation at a +// recorded sequence — nondeterminism leaked between yields. It always +// fails the run loudly. +type Divergence struct { + Sequence int `json:"sequence"` + Expected string `json:"expected_digest"` + Got string `json:"got_digest"` + Detail string `json:"detail,omitempty"` +} + +// JournalEntry is one recorded request/response pair handed to the skill +// subprocess for replay. +type JournalEntry struct { + Request Request `json:"request"` + Response ResponseEnvelope `json:"response"` +} + +// Journal is the replay input: the run identity plus all answered +// operations in sequence order. +type Journal struct { + RunID string `json:"run_id"` + Skill SkillRef `json:"skill"` + Entries []JournalEntry `json:"entries"` +} + +// DigestBytes returns the canonical content digest string for a byte slice. +func DigestBytes(b []byte) string { + sum := sha256.Sum256(b) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// RequestDigest is the canonical digest of a request, used for replay +// divergence detection: kind, id, payload, and output schema all bind. +// JSON fields are compacted first so that a request digests identically +// before and after a round-trip through the log (encoding/json compacts +// embedded RawMessage on marshal). +func RequestDigest(r Request) string { + var buf bytes.Buffer + buf.WriteString(string(r.Kind)) + buf.WriteByte(0) + buf.WriteString(r.ID) + buf.WriteByte(0) + buf.Write(compactJSON(r.Payload)) + buf.WriteByte(0) + buf.Write(compactJSON(r.OutputSchema)) + return DigestBytes(buf.Bytes()) +} + +func compactJSON(raw json.RawMessage) []byte { + if len(raw) == 0 { + return nil + } + var buf bytes.Buffer + if err := json.Compact(&buf, raw); err != nil { + return raw + } + return buf.Bytes() +} + +// DigestSkillDir computes the skill source digest: sha256 over the sorted +// relative paths and contents of *.go, SKILL.md, and go.mod files. +func DigestSkillDir(dir string) (string, error) { + var files []string + err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + name := d.Name() + if name == ".yield" || name == "fixtures" || strings.HasPrefix(name, ".") && path != dir { + return filepath.SkipDir + } + return nil + } + base := d.Name() + if strings.HasSuffix(base, ".go") || base == "SKILL.md" || base == "go.mod" { + files = append(files, path) + } + return nil + }) + if err != nil { + return "", err + } + sort.Strings(files) + h := sha256.New() + for _, f := range files { + rel, err := filepath.Rel(dir, f) + if err != nil { + return "", err + } + b, err := os.ReadFile(f) + if err != nil { + return "", err + } + fmt.Fprintf(h, "%s\x00%d\x00", filepath.ToSlash(rel), len(b)) + h.Write(b) + } + return "sha256:" + hex.EncodeToString(h.Sum(nil)), nil +} + +// ValidateResult checks a completed result against the request's embedded +// JSON schema. A nil schema accepts any JSON value. +func ValidateResult(schema, result json.RawMessage) error { + if len(schema) == 0 { + return nil + } + doc, err := jsonschema.UnmarshalJSON(bytes.NewReader(schema)) + if err != nil { + return fmt.Errorf("output schema is not valid JSON: %w", err) + } + c := jsonschema.NewCompiler() + if err := c.AddResource("schema.json", doc); err != nil { + return err + } + compiled, err := c.Compile("schema.json") + if err != nil { + return fmt.Errorf("output schema does not compile: %w", err) + } + val, err := jsonschema.UnmarshalJSON(bytes.NewReader(result)) + if err != nil { + return fmt.Errorf("result is not valid JSON: %w", err) + } + if err := compiled.Validate(val); err != nil { + return fmt.Errorf("schema-invalid result: %w", err) + } + return nil +} diff --git a/labs/22-yield/yield/internal/protocol/protocol_test.go b/labs/22-yield/yield/internal/protocol/protocol_test.go new file mode 100644 index 00000000..fdb5cbba --- /dev/null +++ b/labs/22-yield/yield/internal/protocol/protocol_test.go @@ -0,0 +1,93 @@ +package protocol + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" +) + +func TestRequestDigestBindsAllFields(t *testing.T) { + base := Request{ID: "a", Kind: OpAskUser, Payload: json.RawMessage(`{"q":1}`)} + same := Request{ID: "a", Kind: OpAskUser, Payload: json.RawMessage(`{"q":1}`)} + if RequestDigest(base) != RequestDigest(same) { + t.Fatal("identical requests must digest identically") + } + for _, changed := range []Request{ + {ID: "b", Kind: OpAskUser, Payload: json.RawMessage(`{"q":1}`)}, + {ID: "a", Kind: OpAgentTask, Payload: json.RawMessage(`{"q":1}`)}, + {ID: "a", Kind: OpAskUser, Payload: json.RawMessage(`{"q":2}`)}, + {ID: "a", Kind: OpAskUser, Payload: json.RawMessage(`{"q":1}`), OutputSchema: json.RawMessage(`{}`)}, + } { + if RequestDigest(base) == RequestDigest(changed) { + t.Fatalf("digest must bind every field; collision on %+v", changed) + } + } +} + +func TestValidateResult(t *testing.T) { + schema := json.RawMessage(`{"type":"object","required":["value"],"properties":{"value":{"type":"string"}}}`) + if err := ValidateResult(schema, json.RawMessage(`{"value":"ok"}`)); err != nil { + t.Fatalf("valid result rejected: %v", err) + } + if err := ValidateResult(schema, json.RawMessage(`{"other":1}`)); err == nil { + t.Fatal("schema-invalid result must be rejected") + } + if err := ValidateResult(nil, json.RawMessage(`{"anything":true}`)); err != nil { + t.Fatalf("nil schema accepts any JSON: %v", err) + } +} + +func TestDigestSkillDirIsContentBound(t *testing.T) { + dir := t.TempDir() + write := func(name, content string) { + if err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0o644); err != nil { + t.Fatal(err) + } + } + write("main.go", "package main") + write("SKILL.md", "prose") + d1, err := DigestSkillDir(dir) + if err != nil { + t.Fatal(err) + } + d2, err := DigestSkillDir(dir) + if err != nil { + t.Fatal(err) + } + if d1 != d2 { + t.Fatal("digest must be deterministic") + } + write("main.go", "package main // changed") + d3, err := DigestSkillDir(dir) + if err != nil { + t.Fatal(err) + } + if d3 == d1 { + t.Fatal("source change must change the digest") + } + // Run state must not affect the digest. + if err := os.MkdirAll(filepath.Join(dir, ".yield", "runs"), 0o755); err != nil { + t.Fatal(err) + } + write(filepath.Join(".yield", "runs", "x.jsonl"), "") + d4, err := DigestSkillDir(dir) + if err != nil { + t.Fatal(err) + } + if d4 != d3 { + t.Fatal("run logs must not affect the skill digest") + } +} + +func TestRequestDigestIsCompactionInvariant(t *testing.T) { + pretty := Request{ID: "a", Kind: OpAgentTask, + Payload: json.RawMessage("{\n \"q\": 1\n}"), + OutputSchema: json.RawMessage("{\n \"type\": \"object\"\n}")} + compact := Request{ID: "a", Kind: OpAgentTask, + Payload: json.RawMessage(`{"q":1}`), + OutputSchema: json.RawMessage(`{"type":"object"}`)} + if RequestDigest(pretty) != RequestDigest(compact) { + t.Fatal("digest must be invariant under JSON compaction (log round-trips compact RawMessage)") + } +} diff --git a/labs/22-yield/yield/internal/runlog/runlog.go b/labs/22-yield/yield/internal/runlog/runlog.go new file mode 100644 index 00000000..b3ca1168 --- /dev/null +++ b/labs/22-yield/yield/internal/runlog/runlog.go @@ -0,0 +1,129 @@ +// Package runlog is the append-only event log that owns every run's state. +// One JSONL file per run; sequence numbers are monotone; nothing is ever +// rewritten. Replaying the log is the only way run state is reconstructed. +package runlog + +import ( + "bufio" + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" +) + +type EventType string + +const ( + RunStarted EventType = "run.started" + OperationRequested EventType = "operation.requested" + OperationCompleted EventType = "operation.completed" + ResponseRejected EventType = "response.rejected" + RequirementPassed EventType = "requirement.passed" + RequirementFailed EventType = "requirement.failed" + DigestMigrated EventType = "digest.migrated" + ReplayDiverged EventType = "replay.diverged" + RunCompleted EventType = "run.completed" + RunBlocked EventType = "run.blocked" + RunRefused EventType = "run.refused" +) + +// Event is one appended fact. Seq is monotone from 1 within a run. +type Event struct { + Seq int `json:"seq"` + Type EventType `json:"type"` + At time.Time `json:"at"` + Data json.RawMessage `json:"data,omitempty"` +} + +// Log is an open run log. Appends go straight to disk with a sync. +type Log struct { + Path string + events []Event +} + +// RunsDir returns the runs directory under a root (usually the skill dir +// or the cwd), creating it if needed. +func RunsDir(root string) (string, error) { + dir := filepath.Join(root, ".yield", "runs") + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", err + } + return dir, nil +} + +// Create starts a new log file for a run. It refuses to overwrite. +func Create(runsDir, runID string) (*Log, error) { + path := filepath.Join(runsDir, runID+".jsonl") + if _, err := os.Stat(path); err == nil { + return nil, fmt.Errorf("run log already exists: %s", path) + } + if err := os.WriteFile(path, nil, 0o644); err != nil { + return nil, err + } + return &Log{Path: path}, nil +} + +// Open loads an existing run log and verifies sequence monotonicity. +func Open(runsDir, runID string) (*Log, error) { + path := filepath.Join(runsDir, runID+".jsonl") + f, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("no such run: %s: %w", runID, err) + } + defer f.Close() + l := &Log{Path: path} + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 1024*1024), 16*1024*1024) + line := 0 + for sc.Scan() { + line++ + if len(sc.Bytes()) == 0 { + continue + } + var e Event + if err := json.Unmarshal(sc.Bytes(), &e); err != nil { + return nil, fmt.Errorf("corrupt run log at line %d: %w", line, err) + } + if e.Seq != len(l.events)+1 { + return nil, fmt.Errorf("run log sequence broken at line %d: got seq %d, want %d", line, e.Seq, len(l.events)+1) + } + l.events = append(l.events, e) + } + if err := sc.Err(); err != nil { + return nil, err + } + return l, nil +} + +// Append writes one event with the next sequence number, fsynced. +func (l *Log) Append(t EventType, data any) (Event, error) { + raw, err := json.Marshal(data) + if err != nil { + return Event{}, err + } + e := Event{Seq: len(l.events) + 1, Type: t, At: time.Now().UTC(), Data: raw} + line, err := json.Marshal(e) + if err != nil { + return Event{}, err + } + f, err := os.OpenFile(l.Path, os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + return Event{}, err + } + defer f.Close() + if _, err := f.Write(append(line, '\n')); err != nil { + return Event{}, err + } + if err := f.Sync(); err != nil { + return Event{}, err + } + l.events = append(l.events, e) + return e, nil +} + +// Events returns the loaded events in order. +func (l *Log) Events() []Event { return l.events } + +// Decode unmarshals an event's data into v. +func (e Event) Decode(v any) error { return json.Unmarshal(e.Data, v) } diff --git a/labs/22-yield/yield/internal/runlog/runlog_test.go b/labs/22-yield/yield/internal/runlog/runlog_test.go new file mode 100644 index 00000000..6fababd6 --- /dev/null +++ b/labs/22-yield/yield/internal/runlog/runlog_test.go @@ -0,0 +1,57 @@ +package runlog + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestAppendOnlyMonotoneSequence(t *testing.T) { + dir := t.TempDir() + l, err := Create(dir, "run_x") + if err != nil { + t.Fatal(err) + } + for i, typ := range []EventType{RunStarted, OperationRequested, OperationCompleted} { + e, err := l.Append(typ, map[string]int{"i": i}) + if err != nil { + t.Fatal(err) + } + if e.Seq != i+1 { + t.Fatalf("want seq %d, got %d", i+1, e.Seq) + } + } + reloaded, err := Open(dir, "run_x") + if err != nil { + t.Fatal(err) + } + if len(reloaded.Events()) != 3 { + t.Fatalf("want 3 events after reload, got %d", len(reloaded.Events())) + } +} + +func TestCreateRefusesOverwrite(t *testing.T) { + dir := t.TempDir() + if _, err := Create(dir, "run_x"); err != nil { + t.Fatal(err) + } + if _, err := Create(dir, "run_x"); err == nil { + t.Fatal("creating over an existing run log must be refused") + } +} + +func TestOpenRefusesBrokenSequence(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "run_bad.jsonl") + lines := `{"seq":1,"type":"run.started","at":"2026-08-01T00:00:00Z"} +{"seq":3,"type":"run.completed","at":"2026-08-01T00:00:01Z"} +` + if err := os.WriteFile(path, []byte(lines), 0o644); err != nil { + t.Fatal(err) + } + _, err := Open(dir, "run_bad") + if err == nil || !strings.Contains(err.Error(), "sequence broken") { + t.Fatalf("want sequence-broken refusal, got %v", err) + } +} diff --git a/labs/22-yield/yield/sdk/yield/yield.go b/labs/22-yield/yield/sdk/yield/yield.go new file mode 100644 index 00000000..2417b75f --- /dev/null +++ b/labs/22-yield/yield/sdk/yield/yield.go @@ -0,0 +1,216 @@ +// Package yield is the skill-program SDK. A skill is an ordinary Go main +// package that calls Main with a deterministic program. Every side effect +// crosses a yielded primitive; code between yields must be deterministic — +// that is what makes replay-based resume sound. +// +// Execution model (deterministic re-execution): on every run/resume the +// program re-executes from the top. Recorded responses are fed back in +// order; at the first unanswered operation the SDK emits a yield.v1 +// request envelope on stdout and exits. If a replayed step produces a +// different operation than the journal recorded, the SDK reports +// divergence and the run fails loudly — it never silently forks. +package yield + +import ( + "encoding/json" + "fmt" + "os" + + "yield/internal/protocol" +) + +// EnvJournal names the environment variable pointing at the journal file +// the supervisor (yskill) hands to the subprocess. +const EnvJournal = "YIELD_JOURNAL" + +// Context carries the replay cursor and the primitives. +type Context struct { + journal protocol.Journal + idx int + requirements []protocol.Requirement +} + +// Outcome is what a program returns on success. +type Outcome struct { + Result any +} + +// Complete finishes the run with a result; evidence is the requirement +// trail accumulated via Require. +func (c *Context) Complete(result any) (Outcome, error) { + return Outcome{Result: result}, nil +} + +// Blocked ends the run at a true frontier, explicitly. +type BlockedError struct{ Reason string } + +func (e *BlockedError) Error() string { return "blocked: " + e.Reason } + +// Refused ends the run because the skill declines to proceed. +type RefusedError struct{ Reason string } + +func (e *RefusedError) Error() string { return "refused: " + e.Reason } + +// Blocked returns the terminal blocked error. +func (c *Context) Blocked(reason string) error { return &BlockedError{Reason: reason} } + +// Refused returns the terminal refused error. +func (c *Context) Refused(reason string) error { return &RefusedError{Reason: reason} } + +// AskUser yields a question to be asked through the host's normal +// interface and returns the selected value on resume. +func (c *Context) AskUser(id, question string, options ...protocol.Option) string { + payload := mustJSON(protocol.AskUserPayload{Question: question, Options: options}) + resp := c.step(protocol.Request{ + ID: id, Kind: protocol.OpAskUser, Payload: payload, + OutputSchema: json.RawMessage(`{"type":"object","required":["value"],"properties":{"value":{"type":"string"}}}`), + }) + var r protocol.AskUserResult + mustDecode(resp.Result, &r) + return r.Value +} + +// AgentTask delegates reasoning to the model. schema (JSON Schema bytes, +// may be nil) is enforced by the supervisor on resume; the returned raw +// message is schema-valid by construction. +func (c *Context) AgentTask(id, instruction string, contextData any, schema json.RawMessage) json.RawMessage { + var ctxRaw json.RawMessage + if contextData != nil { + ctxRaw = mustJSON(contextData) + } + payload := mustJSON(protocol.AgentTaskPayload{Instruction: instruction, Context: ctxRaw}) + resp := c.step(protocol.Request{ID: id, Kind: protocol.OpAgentTask, Payload: payload, OutputSchema: schema}) + return resp.Result +} + +// RunCommand yields a command that yskill executes itself — the result is +// observed fact, not the agent's account of it. +func (c *Context) RunCommand(id, command string, timeoutSeconds int) protocol.CommandResult { + payload := mustJSON(protocol.RunCommandPayload{Command: command, TimeoutSeconds: timeoutSeconds}) + resp := c.step(protocol.Request{ID: id, Kind: protocol.OpRunCommand, Payload: payload}) + var r protocol.CommandResult + mustDecode(resp.Result, &r) + return r +} + +// Require binds a claim to evidence. A failed requirement terminates the +// program immediately with a requirement_failed outcome; completion is +// structurally unreachable past a failed requirement. +func (c *Context) Require(ok bool, claim string, evidence any) { + req := protocol.Requirement{Claim: claim, Passed: ok} + if evidence != nil { + req.EvidenceDigest = protocol.DigestBytes(mustJSON(evidence)) + } + c.requirements = append(c.requirements, req) + if !ok { + emit(protocol.ProgramOutput{ + Type: protocol.OutputTerminal, + Terminal: &protocol.TerminalOutcome{Status: protocol.StatusRequirementFailed, Reason: claim}, + Requirements: c.requirements, + }) + } +} + +// step is the yield mechanism: replay if recorded, emit-and-exit if not, +// diverge loudly if the program no longer matches its own history. +func (c *Context) step(req protocol.Request) protocol.ResponseEnvelope { + seq := c.idx + 1 + if c.idx < len(c.journal.Entries) { + entry := c.journal.Entries[c.idx] + want := protocol.RequestDigest(entry.Request) + got := protocol.RequestDigest(req) + if want != got { + emit(protocol.ProgramOutput{ + Type: protocol.OutputDiverged, + Divergence: &protocol.Divergence{ + Sequence: seq, Expected: want, Got: got, + Detail: fmt.Sprintf("replay produced operation %q (%s) where the journal recorded %q (%s)", req.ID, req.Kind, entry.Request.ID, entry.Request.Kind), + }, + }) + } + c.idx++ + return entry.Response + } + c.idx++ + emit(protocol.ProgramOutput{ + Type: protocol.OutputRequest, + Envelope: &protocol.RequestEnvelope{ + Protocol: protocol.Version, + RunID: c.journal.RunID, + Skill: c.journal.Skill, + Sequence: seq, + Request: req, + }, + Requirements: c.requirements, + }) + panic("unreachable") +} + +// Main runs a skill program under the supervisor protocol. It reads the +// journal named by YIELD_JOURNAL, executes the program, and emits exactly +// one ProgramOutput on stdout. +func Main(program func(*Context) (Outcome, error)) { + path := os.Getenv(EnvJournal) + if path == "" { + fmt.Fprintln(os.Stderr, "yield: YIELD_JOURNAL is not set; this program is run by yskill, not directly") + os.Exit(2) + } + b, err := os.ReadFile(path) + if err != nil { + fmt.Fprintf(os.Stderr, "yield: cannot read journal: %v\n", err) + os.Exit(2) + } + var j protocol.Journal + if err := json.Unmarshal(b, &j); err != nil { + fmt.Fprintf(os.Stderr, "yield: corrupt journal: %v\n", err) + os.Exit(2) + } + ctx := &Context{journal: j} + out, err := program(ctx) + if err != nil { + switch e := err.(type) { + case *BlockedError: + emit(protocol.ProgramOutput{Type: protocol.OutputTerminal, + Terminal: &protocol.TerminalOutcome{Status: protocol.StatusBlocked, Reason: e.Reason}, + Requirements: ctx.requirements}) + case *RefusedError: + emit(protocol.ProgramOutput{Type: protocol.OutputTerminal, + Terminal: &protocol.TerminalOutcome{Status: protocol.StatusRefused, Reason: e.Reason}, + Requirements: ctx.requirements}) + default: + fmt.Fprintf(os.Stderr, "yield: program error: %v\n", err) + os.Exit(1) + } + } + emit(protocol.ProgramOutput{Type: protocol.OutputTerminal, + Terminal: &protocol.TerminalOutcome{Status: protocol.StatusCompleted, Result: mustJSON(out.Result)}, + Requirements: ctx.requirements}) +} + +// emit writes the single ProgramOutput and exits the process. A skill +// execution produces exactly one output. +func emit(out protocol.ProgramOutput) { + enc := json.NewEncoder(os.Stdout) + if err := enc.Encode(out); err != nil { + fmt.Fprintf(os.Stderr, "yield: cannot emit output: %v\n", err) + os.Exit(1) + } + os.Exit(0) +} + +func mustJSON(v any) json.RawMessage { + if raw, ok := v.(json.RawMessage); ok { + return raw + } + b, err := json.Marshal(v) + if err != nil { + panic(fmt.Sprintf("yield: unmarshalable value: %v", err)) + } + return b +} + +func mustDecode(raw json.RawMessage, v any) { + if err := json.Unmarshal(raw, v); err != nil { + panic(fmt.Sprintf("yield: recorded response does not decode: %v", err)) + } +}