diff --git a/.changeset/public-observation-api.md b/.changeset/public-observation-api.md new file mode 100644 index 0000000..102c1d1 --- /dev/null +++ b/.changeset/public-observation-api.md @@ -0,0 +1,9 @@ +--- +"@operatorstack/yield": minor +--- + +Publish the Go receipt projection, verification, durable store, and local +report aggregation as `github.com/operatorstack/yield/observation`. Add strict +typed receipt readers and digest verification to the TypeScript, Python, and +Rust SDKs without changing the observation schema, canonical bytes, storage +layout, or CLI. diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..cc3a21b --- /dev/null +++ b/.gitattributes @@ -0,0 +1 @@ +ir/yield.observation.v1/testdata/*.jsonl text eol=lf diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 7fcfae6..ea7b67b 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -97,7 +97,7 @@ jobs: cd "$smoke_dir" npm init -y >/dev/null npm install "$tarball" >/dev/null - node --input-type=module -e 'import { defineSkill } from "@operatorstack/yield"; if (typeof defineSkill !== "function") process.exit(1)' + node --input-type=module -e 'import { defineSkill, parseRunReceipt } from "@operatorstack/yield"; if (typeof defineSkill !== "function" || typeof parseRunReceipt !== "function") process.exit(1)' python: name: Python SDK diff --git a/cmd/yskill/main.go b/cmd/yskill/main.go index b4ccd50..5094363 100644 --- a/cmd/yskill/main.go +++ b/cmd/yskill/main.go @@ -23,8 +23,8 @@ import ( "github.com/operatorstack/yield/internal/engine" "github.com/operatorstack/yield/internal/outbox" "github.com/operatorstack/yield/internal/protocol" - "github.com/operatorstack/yield/internal/receipt" "github.com/operatorstack/yield/internal/runlog" + receipt "github.com/operatorstack/yield/observation" ) const usage = `yskill — run and resume skill workflows @@ -559,12 +559,12 @@ func cmdReport(args []string) error { if err != nil { return err } - store := receipt.StoreForRunsDir(e.RunsDir) + store := receipt.NewStore(filepath.Dir(e.RunsDir)) ids, err := store.ListRuns() if err != nil { return err } - receipts := make([]*receipt.RunReceipt, 0, len(ids)) + receipts := make([]receipt.RunReceipt, 0, len(ids)) for _, id := range ids { r, _, loadErr := store.LoadRun(id) if loadErr != nil { diff --git a/docs/reference/run-receipts.md b/docs/reference/run-receipts.md index 5fdea54..cdaac93 100644 --- a/docs/reference/run-receipts.md +++ b/docs/reference/run-receipts.md @@ -12,6 +12,63 @@ operation kinds and timing, typed rejection and requirement outcomes, divergence digests, terminal disposition, and optional experiment identifiers. All four SDKs use the same supervisor projection. +## Go API + +Go hosts can project, verify, store, and aggregate receipts without invoking +the CLI. The public package is +[`github.com/operatorstack/yield/observation`](https://pkg.go.dev/github.com/operatorstack/yield/observation). + +```go +receipt, err := observation.Project(journalPrefix) +if err != nil { + return err +} +canonical, err := observation.CanonicalBytes(receipt) +if err != nil { + return err +} + +store := observation.NewStore(yieldDir) +if err := store.Put(receipt, canonical); err != nil { + return err +} +``` + +`Project` accepts one exact, complete `.yield/runs/.jsonl` prefix and +performs no I/O. It rejects partial lines, malformed ordering, unknown event +types, and unknown journal-envelope fields. `Parse` accepts only canonical +receipt bytes whose digest verifies. These checks make the journal prefix the +explicit input and keep receipt files derived, rather than authoritative. + +## SDK readers + +The Go supervisor remains the only receipt projector. TypeScript, Python, and +Rust expose strict readers for the same canonical receipt bytes: + +```typescript +import { parseRunReceipt } from "@operatorstack/yield" + +const receipt = parseRunReceipt(bytes) +``` + +```python +from yieldskill import parse_run_receipt + +receipt = parse_run_receipt(data) +``` + +```rust +use yieldskill::RunReceipt; + +let receipt = RunReceipt::parse_and_verify(bytes)?; +``` + +Each reader exports schema-bound v1 types, rejects unknown fields and invalid +enum values, reproduces Yield's integer-only RFC 8785 canonicalization +profile, and verifies the embedded SHA-256 receipt digest. The SDKs share one +Go-generated golden fixture. They do not read journals, project receipts, +write receipt storage, or participate in replay. + ## Privacy boundary Receipts do not contain prompts, instructions, model responses, user answers, diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 74bad07..1edcc98 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -22,8 +22,8 @@ import ( "github.com/gofrs/flock" "github.com/operatorstack/yield/internal/guard" "github.com/operatorstack/yield/internal/protocol" - "github.com/operatorstack/yield/internal/receipt" "github.com/operatorstack/yield/internal/runlog" + receipt "github.com/operatorstack/yield/observation" ) // Engine binds a skill directory to a runs directory. @@ -393,23 +393,23 @@ func (e *Engine) MaterializeReceipt(runID string) (*receipt.RunReceipt, []byte, if err != nil { return nil, nil, err } - if err := receipt.StoreForRunsDir(e.RunsDir).Put(r, raw); err != nil { + if err := receipt.NewStore(filepath.Dir(e.RunsDir)).Put(*r, raw); err != nil { return nil, nil, err } return r, raw, nil } func (e *Engine) project(runID string) (*receipt.RunReceipt, []byte, error) { - l, rawJournal, err := runlog.OpenSnapshot(e.RunsDir, runID) + _, rawJournal, err := runlog.OpenSnapshot(e.RunsDir, runID) if err != nil { return nil, nil, err } - r, err := receipt.Project(receipt.Snapshot{Bytes: rawJournal, Events: l.Events()}) + r, err := receipt.Project(rawJournal) if err != nil { return nil, nil, err } raw, err := receipt.CanonicalBytes(r) - return r, raw, err + return &r, raw, err } func (e *Engine) materialize(runID string) error { @@ -417,7 +417,7 @@ func (e *Engine) materialize(runID string) error { if err != nil { return err } - return receipt.StoreForRunsDir(e.RunsDir).Put(r, raw) + return receipt.NewStore(filepath.Dir(e.RunsDir)).Put(*r, raw) } // ListRuns returns known run IDs, newest last. diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index adf861a..165d407 100644 --- a/internal/engine/engine_test.go +++ b/internal/engine/engine_test.go @@ -17,8 +17,8 @@ import ( "github.com/operatorstack/yield/internal/guard" "github.com/operatorstack/yield/internal/outbox" "github.com/operatorstack/yield/internal/protocol" - "github.com/operatorstack/yield/internal/receipt" "github.com/operatorstack/yield/internal/runlog" + receipt "github.com/operatorstack/yield/observation" ) // testEngine points at a testdata skill but keeps run logs in a temp dir, @@ -38,7 +38,7 @@ func TestStartRunMaterializesReceiptBeforeReturn(t *testing.T) { if err != nil { t.Fatal(err) } - r, _, err := receipt.StoreForRunsDir(e.RunsDir).LoadRun(p.RunID) + r, _, err := receipt.NewStore(filepath.Dir(e.RunsDir)).LoadRun(p.RunID) if err != nil { t.Fatal(err) } @@ -85,7 +85,7 @@ func TestInitializationFailureHasRunIDJournalAndReceipt(t *testing.T) { if got := l.Events(); len(got) != 2 || got[0].Type != runlog.RunOpened || got[1].Type != runlog.RunInitializationFailed { t.Fatalf("unexpected initialization journal: %+v", got) } - r, _, loadErr := receipt.StoreForRunsDir(e.RunsDir).LoadRun(runErr.RunID) + r, _, loadErr := receipt.NewStore(filepath.Dir(e.RunsDir)).LoadRun(runErr.RunID) if loadErr != nil { t.Fatal(loadErr) } @@ -131,7 +131,7 @@ func main() { fmt.Println("{\"type\":\"terminal\",\"terminal\":{\"status\":\"com if progress.Terminal == nil || progress.Terminal.Status != protocol.StatusCompleted { t.Fatalf("workspace Rust skill did not reach completion: %+v", progress) } - r, _, loadErr := receipt.StoreForRunsDir(runsDir).LoadRun(progress.RunID) + r, _, loadErr := receipt.NewStore(filepath.Dir(runsDir)).LoadRun(progress.RunID) if loadErr != nil { t.Fatal(loadErr) } diff --git a/internal/outbox/outbox.go b/internal/outbox/outbox.go index 2b8c5ae..c02cc8a 100644 --- a/internal/outbox/outbox.go +++ b/internal/outbox/outbox.go @@ -22,7 +22,7 @@ import ( "time" "github.com/gofrs/flock" - "github.com/operatorstack/yield/internal/receipt" + receipt "github.com/operatorstack/yield/observation" ) type Manager struct { @@ -71,7 +71,7 @@ func (m *Manager) Enqueue(sinkID string, r *receipt.RunReceipt, raw []byte) erro if err := r.Validate(); err != nil { return err } - if err := receipt.VerifyCanonical(r, raw); err != nil { + if err := receipt.VerifyCanonical(*r, raw); err != nil { return err } path, err := m.pendingPath(sinkID, r.ReceiptDigest) @@ -551,7 +551,7 @@ func verifyReceipt(raw []byte, digest string) error { if r.ReceiptDigest != digest { return fmt.Errorf("outbox: pending receipt digest does not match its name") } - return receipt.VerifyCanonical(&r, raw) + return receipt.VerifyCanonical(r, raw) } type digestWriter struct { diff --git a/internal/outbox/outbox_test.go b/internal/outbox/outbox_test.go index 17592b9..06b43ce 100644 --- a/internal/outbox/outbox_test.go +++ b/internal/outbox/outbox_test.go @@ -11,29 +11,36 @@ import ( "testing" "time" - "github.com/operatorstack/yield/internal/receipt" + "github.com/operatorstack/yield/internal/protocol" + "github.com/operatorstack/yield/internal/runlog" + receipt "github.com/operatorstack/yield/observation" ) func testReceipt(t *testing.T) (*receipt.RunReceipt, []byte) { t.Helper() - r := &receipt.RunReceipt{ - Schema: receipt.Schema, Kind: receipt.Kind, - Journal: receipt.JournalBinding{RunID: "run_1", HeadSequence: 1, HeadDigest: digestBytes([]byte("journal"))}, - Run: receipt.RunIdentity{ID: "run_1"}, - Skill: receipt.SkillIdentity{Name: "test", BindingDigest: digestBytes([]byte("skill"))}, - Timing: receipt.TimingSummary{StartedAt: "2026-08-20T10:00:00Z", LastObservedAt: "2026-08-20T10:00:00Z"}, - Operations: []receipt.OperationObservation{}, OperationSummaries: []receipt.OperationSummary{}, - Outcome: receipt.OutcomeSummary{Phase: "advancing"}, Requirements: []receipt.RequirementOutcome{}, - ResponseRejections: []receipt.ResponseRejectionSummary{}, Divergences: []receipt.DivergenceOutcome{}, + event := runlog.Event{ + Seq: 1, + Type: runlog.RunStarted, + At: time.Date(2026, 8, 20, 10, 0, 0, 0, time.UTC), + } + event.Data, _ = json.Marshal(map[string]any{ + "run_id": "run_1", + "skill": protocol.SkillRef{Name: "test", Digest: digestBytes([]byte("skill"))}, + }) + journal, err := json.Marshal(event) + if err != nil { + t.Fatal(err) } - if err := receipt.Seal(r); err != nil { + journal = append(journal, '\n') + r, err := receipt.Project(journal) + if err != nil { t.Fatal(err) } raw, err := receipt.CanonicalBytes(r) if err != nil { t.Fatal(err) } - return r, raw + return &r, raw } func TestEnqueueIsByteIdempotent(t *testing.T) { diff --git a/internal/runlog/runlog.go b/internal/runlog/runlog.go index 8fdde63..50cb553 100644 --- a/internal/runlog/runlog.go +++ b/internal/runlog/runlog.go @@ -9,6 +9,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "os" "path/filepath" "runtime" @@ -113,6 +114,46 @@ func OpenSnapshot(runsDir, runID string) (*Log, []byte, error) { return l, raw, nil } +// ParseSnapshot verifies and decodes one exact, complete JSONL journal prefix. +func ParseSnapshot(raw []byte) ([]Event, error) { + if len(raw) == 0 { + return nil, fmt.Errorf("run log is empty") + } + if raw[len(raw)-1] != '\n' { + return nil, fmt.Errorf("run log ends with a partial event") + } + var events []Event + sc := bufio.NewScanner(bytes.NewReader(raw)) + sc.Buffer(make([]byte, 0, 1024*1024), 16*1024*1024) + line := 0 + for sc.Scan() { + line++ + if len(sc.Bytes()) == 0 { + return nil, fmt.Errorf("corrupt run log at line %d: blank event", line) + } + decoder := json.NewDecoder(bytes.NewReader(sc.Bytes())) + decoder.DisallowUnknownFields() + var event Event + if err := decoder.Decode(&event); err != nil { + return nil, fmt.Errorf("corrupt run log at line %d: %w", line, err) + } + if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { + if err == nil { + err = fmt.Errorf("multiple JSON values") + } + return nil, fmt.Errorf("corrupt run log at line %d: trailing content: %w", line, err) + } + if event.Seq != len(events)+1 { + return nil, fmt.Errorf("run log sequence broken at line %d: got seq %d, want %d", line, event.Seq, len(events)+1) + } + events = append(events, event) + } + if err := sc.Err(); err != nil { + return nil, err + } + return events, nil +} + func parse(path string, raw []byte) (*Log, error) { l := &Log{Path: path} sc := bufio.NewScanner(bytes.NewReader(raw)) diff --git a/ir/README.md b/ir/README.md index 3cd9446..32662da 100644 --- a/ir/README.md +++ b/ir/README.md @@ -6,8 +6,13 @@ This directory contains Yield's language-neutral schemas. - `yield.observation.v1` is the portable observation boundary projected by the Go supervisor from an append-only run journal. SDKs do not implement it. -The Go reference types and schema tests keep both boundaries aligned with the -runtime. +The public Go package +[`observation`](https://pkg.go.dev/github.com/operatorstack/yield/observation) +provides the reference receipt types, pure journal projection, strict parsing, +durable local store, and report aggregation. Schema tests keep that package +aligned with the language-neutral boundary. The TypeScript, Python, and Rust +SDKs provide strict typed readers and digest verification over the same +Go-generated canonical fixture; they do not duplicate projection or storage. ## Files diff --git a/ir/yield.observation.v1/testdata/run-receipt.canonical.jsonl b/ir/yield.observation.v1/testdata/run-receipt.canonical.jsonl new file mode 100644 index 0000000..fd2be7f --- /dev/null +++ b/ir/yield.observation.v1/testdata/run-receipt.canonical.jsonl @@ -0,0 +1 @@ +{"divergences":[],"experiment":{"baseline_variant_id":"baseline-a","cohort_id":"cohort-a","experiment_id":"exp-1","parent_skill_version":"1.2.2","role":"candidate","variant_id":"candidate-a"},"journal":{"head_digest":"sha256:bac3f7194c5baf064fc036f50651123dc882e2672f8de90c438b5160118ddc0e","head_sequence":7,"run_id":"run_fixture"},"kind":"run_receipt","operation_summaries":[{"completed":1,"kind":"agent_task","requested":1,"total_elapsed_ms":3000}],"operations":[{"completed_at":"2026-08-20T10:00:05Z","elapsed_ms":3000,"kind":"agent_task","operation_key_digest":"sha256:05c048343bf1e65d79dc8c423ca8906d6572ada664fbf29b3118aad48849f900","requested_at":"2026-08-20T10:00:02Z","result_digest":"sha256:4444444444444444444444444444444444444444444444444444444444444444","sequence":1}],"outcome":{"phase":"terminal","result_digest":"sha256:4062edaf750fb8074e7e83e0c9028c94e32468a8b6f1614774328ef045150f93","terminal_cause":"completed","terminal_disposition":"completed"},"receipt_digest":"sha256:a0995d74d51e472f61dc2001f82b93fd8086f9c2f2816a22bb7e5c7fa3c01cdd","requirements":[{"claim_digest":"sha256:b8cb5510ac0e4abd6b1ca23d6d2cb6902dd6cd3d4078a061ef65464f5a431cb0","evidence_digest":"sha256:5555555555555555555555555555555555555555555555555555555555555555","outcome":"passed"}],"response_rejections":[{"count":1,"reason":"duplicate-response"}],"run":{"id":"run_fixture","input_digest":"sha256:1111111111111111111111111111111111111111111111111111111111111111"},"runtime":{"compatible":true,"required_version":"0.7.0","supervisor_version":"0.7.0"},"schema":"yield.observation.v1","skill":{"binding_digest":"sha256:2222222222222222222222222222222222222222222222222222222222222222","name":"fixture","source_digest":{"profile":"yield.skill-source.v1","value":"sha256:3333333333333333333333333333333333333333333333333333333333333333"},"version":"1.2.3"},"timing":{"elapsed_ms":8000,"ended_at":"2026-08-20T10:00:08Z","last_observed_at":"2026-08-20T10:00:08Z","started_at":"2026-08-20T10:00:00Z"}} diff --git a/observation/golden_test.go b/observation/golden_test.go new file mode 100644 index 0000000..b2d5931 --- /dev/null +++ b/observation/golden_test.go @@ -0,0 +1,30 @@ +package observation + +import ( + "bytes" + "os" + "path/filepath" + "testing" +) + +func TestSharedCanonicalReceiptFixture(t *testing.T) { + raw, err := os.ReadFile(filepath.Join("..", "ir", "yield.observation.v1", "testdata", "run-receipt.canonical.jsonl")) + if err != nil { + t.Fatal(err) + } + canonical := bytes.TrimSuffix(raw, []byte("\n")) + if len(canonical) == len(raw) || bytes.Contains(canonical, []byte("\n")) { + t.Fatal("shared receipt fixture must be one canonical JSON line followed by one file newline") + } + receipt, err := Parse(canonical) + if err != nil { + t.Fatal(err) + } + encoded, err := CanonicalBytes(receipt) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(encoded, canonical) { + t.Fatal("shared receipt fixture differs from Go canonical bytes") + } +} diff --git a/internal/receipt/ir_test.go b/observation/ir_test.go similarity index 93% rename from internal/receipt/ir_test.go rename to observation/ir_test.go index 8aea732..8562979 100644 --- a/internal/receipt/ir_test.go +++ b/observation/ir_test.go @@ -1,4 +1,4 @@ -package receipt +package observation import ( "bytes" @@ -15,7 +15,7 @@ import ( func observationSchema(t *testing.T) *jsonschema.Schema { t.Helper() - path := filepath.Join("..", "..", "ir", "yield.observation.v1", "run-receipt.schema.json") + path := filepath.Join("..", "ir", "yield.observation.v1", "run-receipt.schema.json") raw, err := os.ReadFile(path) if err != nil { t.Fatal(err) @@ -44,7 +44,7 @@ func TestGoReceiptValidatesAgainstObservationIR(t *testing.T) { event(t, 2, runlog.RunStarted, t0, map[string]any{"run_id": "run_schema", "skill": skill, "input_digest": digest, "supervisor_version": "1.0.0", "required_yield_version": "1.0.0", "source_digest_profile": protocol.SkillSourceProfileV1, "source_digest": digest}), event(t, 3, runlog.RunCompleted, t0.Add(time.Second), map[string]any{"result": json.RawMessage(`{"ok":true}`)}), } - r, err := Project(Snapshot{Events: events, Bytes: journalBytes(t, events)}) + r, err := Project(journalBytes(t, events)) if err != nil { t.Fatal(err) } diff --git a/observation/public_api_test.go b/observation/public_api_test.go new file mode 100644 index 0000000..9d70a50 --- /dev/null +++ b/observation/public_api_test.go @@ -0,0 +1,39 @@ +package observation_test + +import ( + "bytes" + "testing" + + "github.com/operatorstack/yield/observation" +) + +func TestPublicProjectionVerificationAndStore(t *testing.T) { + journal := []byte(`{"seq":1,"type":"run.started","at":"2026-08-20T10:00:00Z","data":{"run_id":"run_public","skill":{"name":"public","digest":"sha256:0000000000000000000000000000000000000000000000000000000000000000"},"input_digest":"sha256:1111111111111111111111111111111111111111111111111111111111111111"}}` + "\n") + receipt, err := observation.Project(journal) + if err != nil { + t.Fatal(err) + } + canonical, err := observation.CanonicalBytes(receipt) + if err != nil { + t.Fatal(err) + } + parsed, err := observation.Parse(canonical) + if err != nil { + t.Fatal(err) + } + if err := observation.VerifyCanonical(parsed, canonical); err != nil { + t.Fatal(err) + } + + store := observation.NewStore(t.TempDir()) + if err := store.Put(parsed, canonical); err != nil { + t.Fatal(err) + } + loaded, loadedBytes, err := store.LoadRun("run_public") + if err != nil { + t.Fatal(err) + } + if loaded.ReceiptDigest != receipt.ReceiptDigest || !bytes.Equal(loadedBytes, canonical) { + t.Fatal("public store round trip changed the receipt") + } +} diff --git a/internal/receipt/receipt.go b/observation/receipt.go similarity index 87% rename from internal/receipt/receipt.go rename to observation/receipt.go index f384a18..623ed00 100644 --- a/internal/receipt/receipt.go +++ b/observation/receipt.go @@ -1,6 +1,6 @@ -// Package receipt projects privacy-safe, portable observations from Yield's +// Package observation projects privacy-safe, portable observations from Yield's // authoritative append-only run journal. -package receipt +package observation import ( "bytes" @@ -8,6 +8,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "io" "sort" "strconv" "strings" @@ -24,6 +25,7 @@ const ( Kind = "run_receipt" ) +// RunReceipt is the canonical privacy-safe projection of one journal prefix. type RunReceipt struct { Schema string `json:"schema"` Kind string `json:"kind"` @@ -42,17 +44,29 @@ type RunReceipt struct { Experiment *ExperimentContext `json:"experiment,omitempty"` } +// OperationKind is one supervisor-observed Yield operation kind. +type OperationKind string + +const ( + OperationAskUser OperationKind = "ask_user" + OperationAgentTask OperationKind = "agent_task" + OperationRunCommand OperationKind = "run_command" +) + +// JournalBinding identifies the exact authoritative journal prefix projected. type JournalBinding struct { RunID string `json:"run_id"` HeadSequence int `json:"head_sequence"` HeadDigest string `json:"head_digest"` } +// RunIdentity identifies the observed run and its privacy-safe input digest. type RunIdentity struct { ID string `json:"id"` InputDigest string `json:"input_digest,omitempty"` } +// SkillIdentity records the skill facts available in the journal. type SkillIdentity struct { Name string `json:"name"` Version string `json:"version,omitempty"` @@ -60,17 +74,20 @@ type SkillIdentity struct { SourceDigest *ProfileDigest `json:"source_digest,omitempty"` } +// ProfileDigest names a versioned digest profile and its value. type ProfileDigest struct { Profile string `json:"profile"` Value string `json:"value"` } +// RuntimeIdentity records authoritative supervisor compatibility facts. type RuntimeIdentity struct { SupervisorVersion string `json:"supervisor_version,omitempty"` RequiredVersion string `json:"required_version,omitempty"` Compatible *bool `json:"compatible,omitempty"` } +// TimingSummary contains journal timestamps and safe derived timing. type TimingSummary struct { StartedAt string `json:"started_at"` LastObservedAt string `json:"last_observed_at"` @@ -79,24 +96,27 @@ type TimingSummary struct { ClockAnomaly bool `json:"clock_anomaly,omitempty"` } +// OperationObservation describes one supervisor-observed operation. type OperationObservation struct { - Sequence int `json:"sequence"` - Kind protocol.OpKind `json:"kind"` - OperationKeyDigest string `json:"operation_key_digest"` - RequestedAt string `json:"requested_at"` - CompletedAt string `json:"completed_at,omitempty"` - ElapsedMS *int64 `json:"elapsed_ms,omitempty"` - ResultDigest string `json:"result_digest,omitempty"` - ClockAnomaly bool `json:"clock_anomaly,omitempty"` + Sequence int `json:"sequence"` + Kind OperationKind `json:"kind"` + OperationKeyDigest string `json:"operation_key_digest"` + RequestedAt string `json:"requested_at"` + CompletedAt string `json:"completed_at,omitempty"` + ElapsedMS *int64 `json:"elapsed_ms,omitempty"` + ResultDigest string `json:"result_digest,omitempty"` + ClockAnomaly bool `json:"clock_anomaly,omitempty"` } +// OperationSummary aggregates observations of one operation kind. type OperationSummary struct { - Kind protocol.OpKind `json:"kind"` - Requested int `json:"requested"` - Completed int `json:"completed"` - TotalElapsedMS int64 `json:"total_elapsed_ms"` + Kind OperationKind `json:"kind"` + Requested int `json:"requested"` + Completed int `json:"completed"` + TotalElapsedMS int64 `json:"total_elapsed_ms"` } +// OutcomeSummary classifies the latest lifecycle and terminal outcome. type OutcomeSummary struct { Phase string `json:"phase"` TerminalDisposition string `json:"terminal_disposition,omitempty"` @@ -105,23 +125,27 @@ type OutcomeSummary struct { FailureCode string `json:"failure_code,omitempty"` } +// RequirementOutcome records a requirement result without its raw claim. type RequirementOutcome struct { Outcome string `json:"outcome"` ClaimDigest string `json:"claim_digest"` EvidenceDigest string `json:"evidence_digest,omitempty"` } +// ResponseRejectionSummary counts one closed response-rejection reason. type ResponseRejectionSummary struct { Reason string `json:"reason"` Count int `json:"count"` } +// DivergenceOutcome records expected and actual digests at a replay sequence. type DivergenceOutcome struct { Sequence int `json:"sequence"` Expected string `json:"expected_digest"` Got string `json:"got_digest"` } +// ExperimentContext groups a run for external evaluation without selecting a winner. type ExperimentContext struct { ExperimentID string `json:"experiment_id"` CohortID string `json:"cohort_id,omitempty"` @@ -131,8 +155,7 @@ type ExperimentContext struct { ParentSkillVersion string `json:"parent_skill_version,omitempty"` } -// Snapshot is the complete, immutable input to one projection. -type Snapshot struct { +type snapshot struct { Bytes []byte Events []runlog.Event } @@ -163,8 +186,22 @@ type operationCompletedData struct { ResultDigest string `json:"result_digest"` } -// Project deterministically derives one receipt from one exact journal prefix. -func Project(snapshot Snapshot) (*RunReceipt, error) { +// Project deterministically derives and seals one receipt from an exact +// append-only run-journal prefix. The prefix must end at a complete JSONL +// record and is never rewritten. +func Project(journalPrefix []byte) (RunReceipt, error) { + events, err := runlog.ParseSnapshot(journalPrefix) + if err != nil { + return RunReceipt{}, err + } + projected, err := project(snapshot{Bytes: journalPrefix, Events: events}) + if err != nil { + return RunReceipt{}, err + } + return *projected, nil +} + +func project(snapshot snapshot) (*RunReceipt, error) { if len(snapshot.Events) == 0 { return nil, fmt.Errorf("receipt: journal is empty") } @@ -297,7 +334,7 @@ func Project(snapshot Snapshot) (*RunReceipt, error) { } operation := &OperationObservation{ Sequence: envelope.Sequence, - Kind: envelope.Request.Kind, + Kind: OperationKind(envelope.Request.Kind), OperationKeyDigest: digest("yield.operation.v1", []byte(string(envelope.Request.Kind)+"\x00"+envelope.Request.ID)), RequestedAt: formatTime(event.At), } @@ -479,7 +516,7 @@ func Project(snapshot Snapshot) (*RunReceipt, error) { sequences = append(sequences, sequence) } sort.Ints(sequences) - summaries := map[protocol.OpKind]*OperationSummary{} + summaries := map[OperationKind]*OperationSummary{} for _, sequence := range sequences { operation := operations[sequence] r.Operations = append(r.Operations, *operation) @@ -496,7 +533,7 @@ func Project(snapshot Snapshot) (*RunReceipt, error) { summary.TotalElapsedMS += *operation.ElapsedMS } } - for _, kind := range []protocol.OpKind{protocol.OpAskUser, protocol.OpAgentTask, protocol.OpRunCommand} { + for _, kind := range []OperationKind{OperationAskUser, OperationAgentTask, OperationRunCommand} { if summary := summaries[kind]; summary != nil { r.OperationSummaries = append(r.OperationSummaries, *summary) } @@ -534,7 +571,7 @@ func Project(snapshot Snapshot) (*RunReceipt, error) { } } - if err := Seal(r); err != nil { + if err := seal(r); err != nil { return nil, err } return r, nil @@ -588,13 +625,12 @@ func digest(domain string, value []byte) string { return "sha256:" + hex.EncodeToString(h.Sum(nil)) } -// Seal sets the deterministic receipt digest. -func Seal(receipt *RunReceipt) error { +func seal(receipt *RunReceipt) error { receipt.ReceiptDigest = "" if err := receipt.Validate(); err != nil { return err } - body, err := CanonicalBytes(receipt) + body, err := CanonicalBytes(*receipt) if err != nil { return err } @@ -604,19 +640,19 @@ func Seal(receipt *RunReceipt) error { // VerifyCanonical proves that bytes are the canonical encoding named by the // receipt digest. -func VerifyCanonical(receipt *RunReceipt, raw []byte) error { - if receipt == nil || receipt.ReceiptDigest == "" { +func VerifyCanonical(receipt RunReceipt, raw []byte) error { + if receipt.ReceiptDigest == "" { return fmt.Errorf("receipt: missing receipt digest") } want := receipt.ReceiptDigest - copy := *receipt - if err := Seal(©); err != nil { + copy := receipt + if err := seal(©); err != nil { return err } if copy.ReceiptDigest != want { return fmt.Errorf("receipt: digest verification failed") } - canonical, err := CanonicalBytes(©) + canonical, err := CanonicalBytes(copy) if err != nil { return err } @@ -626,6 +662,26 @@ func VerifyCanonical(receipt *RunReceipt, raw []byte) error { return nil } +// Parse strictly decodes and verifies one canonical RunReceipt. +func Parse(raw []byte) (RunReceipt, error) { + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.DisallowUnknownFields() + var receipt RunReceipt + if err := decoder.Decode(&receipt); err != nil { + return RunReceipt{}, fmt.Errorf("receipt: decode: %w", err) + } + if err := decoder.Decode(&struct{}{}); err != io.EOF { + if err == nil { + err = fmt.Errorf("multiple JSON values") + } + return RunReceipt{}, fmt.Errorf("receipt: trailing content: %w", err) + } + if err := VerifyCanonical(receipt, raw); err != nil { + return RunReceipt{}, err + } + return receipt, nil +} + // Validate checks the closed Go representation against the public contract. func (receipt *RunReceipt) Validate() error { if receipt == nil || receipt.Schema != Schema || receipt.Kind != Kind { @@ -741,8 +797,8 @@ func (receipt *RunReceipt) Validate() error { return nil } -func validOperationKind(kind protocol.OpKind) bool { - return kind == protocol.OpAskUser || kind == protocol.OpAgentTask || kind == protocol.OpRunCommand +func validOperationKind(kind OperationKind) bool { + return kind == OperationAskUser || kind == OperationAgentTask || kind == OperationRunCommand } func validDigest(value string) bool { @@ -755,7 +811,11 @@ func validDigest(value string) bool { // CanonicalBytes returns RFC 8785-compatible JSON for the receipt's closed, // integer-only data model. -func CanonicalBytes(value any) ([]byte, error) { +func CanonicalBytes(receipt RunReceipt) ([]byte, error) { + return canonicalBytes(receipt) +} + +func canonicalBytes(value any) ([]byte, error) { raw, err := json.Marshal(value) if err != nil { return nil, err diff --git a/internal/receipt/receipt_test.go b/observation/receipt_test.go similarity index 76% rename from internal/receipt/receipt_test.go rename to observation/receipt_test.go index 9579a1b..fe81b81 100644 --- a/internal/receipt/receipt_test.go +++ b/observation/receipt_test.go @@ -1,4 +1,4 @@ -package receipt +package observation import ( "bytes" @@ -27,7 +27,7 @@ func TestProjectCompletedReceiptIsDeterministicAndPrivate(t *testing.T) { event(t, 4, runlog.RequirementPassed, t0.Add(5*time.Second), protocol.Requirement{Claim: secret, Passed: true, EvidenceDigest: protocol.DigestBytes(result)}), event(t, 5, runlog.RunCompleted, t0.Add(6*time.Second), map[string]any{"result": result, "requirements": 1}), } - snapshot := Snapshot{Events: events, Bytes: journalBytes(t, events)} + snapshot := journalBytes(t, events) first, err := Project(snapshot) if err != nil { t.Fatal(err) @@ -58,7 +58,7 @@ func TestProjectLegacyJournalDoesNotInventSourceOrRuntime(t *testing.T) { events := []runlog.Event{ event(t, 1, runlog.RunStarted, t0, map[string]any{"run_id": "run_legacy", "skill": skill, "input_digest": protocol.DigestBytes(nil)}), } - receipt, err := Project(Snapshot{Events: events, Bytes: journalBytes(t, events)}) + receipt, err := Project(journalBytes(t, events)) if err != nil { t.Fatal(err) } @@ -79,7 +79,7 @@ func TestProjectClockAnomalyOmitsDuration(t *testing.T) { event(t, 2, runlog.OperationRequested, t0.Add(time.Second), envelope), event(t, 3, runlog.OperationCompleted, t0, map[string]any{"sequence": 1, "request_id": "ask", "result": json.RawMessage(`{"value":"yes"}`)}), } - receipt, err := Project(Snapshot{Events: events, Bytes: journalBytes(t, events)}) + receipt, err := Project(journalBytes(t, events)) if err != nil { t.Fatal(err) } @@ -89,7 +89,7 @@ func TestProjectClockAnomalyOmitsDuration(t *testing.T) { } func TestCanonicalBytesUsesJCSStringAndKeyRules(t *testing.T) { - raw, err := CanonicalBytes(map[string]any{"😀": "\u2028", "€": "<", "\r": "\n"}) + raw, err := canonicalBytes(map[string]any{"😀": "\u2028", "€": "<", "\r": "\n"}) if err != nil { t.Fatal(err) } @@ -150,7 +150,7 @@ func TestProjectLifecycleClassifications(t *testing.T) { } for _, test := range tests { t.Run(test.name, func(t *testing.T) { - r, err := Project(Snapshot{Events: test.events, Bytes: journalBytes(t, test.events)}) + r, err := Project(journalBytes(t, test.events)) if err != nil { t.Fatal(err) } @@ -173,7 +173,7 @@ func TestProjectRejectsBrokenOperationPairing(t *testing.T) { event(t, 1, runlog.RunStarted, t0, map[string]any{"run_id": "run_broken", "skill": skill}), event(t, 2, runlog.OperationCompleted, t0.Add(time.Second), map[string]any{"sequence": 1, "request_id": "missing", "result_digest": digest}), } - if _, err := Project(Snapshot{Events: events, Bytes: journalBytes(t, events)}); err == nil { + if _, err := Project(journalBytes(t, events)); err == nil { t.Fatal("broken operation pairing was accepted") } } @@ -184,7 +184,7 @@ func TestProjectFailsClosedOnUnknownAndPostTerminalEvents(t *testing.T) { skill := protocol.SkillRef{Name: "closed", Digest: digest} started := event(t, 1, runlog.RunStarted, t0, map[string]any{"run_id": "run_closed", "skill": skill}) unknown := []runlog.Event{started, event(t, 2, runlog.EventType("future.event"), t0.Add(time.Second), map[string]any{})} - if _, err := Project(Snapshot{Events: unknown, Bytes: journalBytes(t, unknown)}); err == nil { + if _, err := Project(journalBytes(t, unknown)); err == nil { t.Fatal("unknown event was silently omitted") } postTerminal := []runlog.Event{ @@ -192,11 +192,63 @@ func TestProjectFailsClosedOnUnknownAndPostTerminalEvents(t *testing.T) { event(t, 2, runlog.RunRefused, t0.Add(time.Second), map[string]any{}), event(t, 3, runlog.RunCompleted, t0.Add(2*time.Second), map[string]any{}), } - if _, err := Project(Snapshot{Events: postTerminal, Bytes: journalBytes(t, postTerminal)}); err == nil { + if _, err := Project(journalBytes(t, postTerminal)); err == nil { t.Fatal("event after terminal was accepted") } } +func TestProjectRejectsPartialBlankAndUnknownEnvelopeFields(t *testing.T) { + t0 := time.Date(2026, 8, 20, 10, 0, 0, 0, time.UTC) + skill := protocol.SkillRef{Name: "strict", Digest: protocol.DigestBytes([]byte("strict"))} + events := []runlog.Event{event(t, 1, runlog.RunStarted, t0, map[string]any{"run_id": "run_strict", "skill": skill})} + complete := journalBytes(t, events) + + for name, journal := range map[string][]byte{ + "partial line": complete[:len(complete)-1], + "blank line": append(append([]byte{}, complete...), '\n'), + "unknown envelope field": []byte(strings.Replace(string(complete), `"data":`, `"unknown":true,"data":`, 1)), + } { + t.Run(name, func(t *testing.T) { + if _, err := Project(journal); err == nil { + t.Fatal("malformed journal prefix was accepted") + } + }) + } +} + +func TestParseRequiresExactCanonicalVerifiedBytes(t *testing.T) { + t0 := time.Date(2026, 8, 20, 10, 0, 0, 0, time.UTC) + skill := protocol.SkillRef{Name: "parse", Digest: protocol.DigestBytes([]byte("parse"))} + events := []runlog.Event{event(t, 1, runlog.RunStarted, t0, map[string]any{"run_id": "run_parse", "skill": skill})} + receipt, err := Project(journalBytes(t, events)) + if err != nil { + t.Fatal(err) + } + canonical, err := CanonicalBytes(receipt) + if err != nil { + t.Fatal(err) + } + parsed, err := Parse(canonical) + if err != nil || parsed.ReceiptDigest != receipt.ReceiptDigest { + t.Fatalf("canonical receipt did not round trip: %+v, %v", parsed, err) + } + + mutatedDigest := bytes.Replace(canonical, []byte(receipt.ReceiptDigest), []byte(protocol.DigestBytes([]byte("wrong"))), 1) + unknownField := append(append([]byte{}, canonical[:len(canonical)-1]...), []byte(`,"prompt":"secret"}`)...) + for name, raw := range map[string][]byte{ + "noncanonical whitespace": append([]byte(" "), canonical...), + "digest mismatch": mutatedDigest, + "unknown field": unknownField, + "trailing content": append(append([]byte{}, canonical...), []byte("\n{}")...), + } { + t.Run(name, func(t *testing.T) { + if _, err := Parse(raw); err == nil { + t.Fatal("invalid receipt bytes were accepted") + } + }) + } +} + func event(t *testing.T, sequence int, kind runlog.EventType, at time.Time, data any) runlog.Event { t.Helper() raw, err := json.Marshal(data) diff --git a/internal/receipt/report.go b/observation/report.go similarity index 84% rename from internal/receipt/report.go rename to observation/report.go index df67fa8..e81566a 100644 --- a/internal/receipt/report.go +++ b/observation/report.go @@ -1,13 +1,12 @@ -package receipt +package observation import ( "fmt" "sort" "time" - - "github.com/operatorstack/yield/internal/protocol" ) +// ReportOptions bounds deterministic local receipt aggregation. type ReportOptions struct { From time.Time To time.Time @@ -16,6 +15,7 @@ type ReportOptions struct { ExperimentID string } +// LocalReport is a descriptive aggregation with no causal or activation claim. type LocalReport struct { ReceiptCount int `json:"receipt_count"` Lifecycle []NamedCount `json:"lifecycle"` @@ -30,19 +30,22 @@ type LocalReport struct { OpenOlderThanThreshold int `json:"open_older_than_threshold,omitempty"` } +// NamedCount is one stable, sorted report group. type NamedCount struct { Name string `json:"name"` Count int `json:"count"` } +// ReportOperationSummary aggregates timing and counts for an operation kind. type ReportOperationSummary struct { - Kind protocol.OpKind `json:"kind"` - Requested int `json:"requested"` - Completed int `json:"completed"` - TotalElapsedMS int64 `json:"total_elapsed_ms"` + Kind OperationKind `json:"kind"` + Requested int `json:"requested"` + Completed int `json:"completed"` + TotalElapsedMS int64 `json:"total_elapsed_ms"` } -func BuildReport(receipts []*RunReceipt, options ReportOptions) (LocalReport, error) { +// BuildReport deterministically aggregates an explicit receipt set and options. +func BuildReport(receipts []RunReceipt, options ReportOptions) (LocalReport, error) { report := LocalReport{ Lifecycle: []NamedCount{}, Terminal: []NamedCount{}, Operations: []ReportOperationSummary{}, ResponseRejections: []NamedCount{}, Requirements: []NamedCount{}, RuntimeGroups: []NamedCount{}, @@ -55,7 +58,7 @@ func BuildReport(receipts []*RunReceipt, options ReportOptions) (LocalReport, er runtimes := map[string]int{} sources := map[string]int{} experiments := map[string]int{} - operations := map[protocol.OpKind]*ReportOperationSummary{} + operations := map[OperationKind]*ReportOperationSummary{} for _, r := range receipts { started, err := time.Parse(time.RFC3339Nano, r.Timing.StartedAt) if err != nil { @@ -115,7 +118,7 @@ func BuildReport(receipts []*RunReceipt, options ReportOptions) (LocalReport, er report.RuntimeGroups = namedCounts(runtimes) report.SourceGroups = namedCounts(sources) report.ExperimentGroups = namedCounts(experiments) - for _, kind := range []protocol.OpKind{protocol.OpAskUser, protocol.OpAgentTask, protocol.OpRunCommand} { + for _, kind := range []OperationKind{OperationAskUser, OperationAgentTask, OperationRunCommand} { if summary := operations[kind]; summary != nil { report.Operations = append(report.Operations, *summary) } diff --git a/internal/receipt/report_test.go b/observation/report_test.go similarity index 76% rename from internal/receipt/report_test.go rename to observation/report_test.go index c323093..bf54cf6 100644 --- a/internal/receipt/report_test.go +++ b/observation/report_test.go @@ -1,29 +1,27 @@ -package receipt +package observation import ( "reflect" "testing" "time" - - "github.com/operatorstack/yield/internal/protocol" ) func TestBuildReportIsDeterministicAndMakesNoCausalClaim(t *testing.T) { - r := &RunReceipt{ + r := RunReceipt{ Run: RunIdentity{ID: "run_1"}, Timing: TimingSummary{StartedAt: "2026-08-20T10:00:00Z"}, Outcome: OutcomeSummary{Phase: "awaiting_response"}, - OperationSummaries: []OperationSummary{{Kind: protocol.OpAskUser, Requested: 1}}, + OperationSummaries: []OperationSummary{{Kind: OperationAskUser, Requested: 1}}, Experiment: &ExperimentContext{ExperimentID: "exp-1", VariantID: "candidate", Role: "candidate"}, } options := ReportOptions{ From: time.Date(2026, 8, 20, 0, 0, 0, 0, time.UTC), To: time.Date(2026, 8, 21, 0, 0, 0, 0, time.UTC), OpenAgeThreshold: time.Hour, ReferenceTime: time.Date(2026, 8, 20, 12, 0, 0, 0, time.UTC), } - first, err := BuildReport([]*RunReceipt{r}, options) + first, err := BuildReport([]RunReceipt{r}, options) if err != nil { t.Fatal(err) } - second, err := BuildReport([]*RunReceipt{r}, options) + second, err := BuildReport([]RunReceipt{r}, options) if err != nil { t.Fatal(err) } diff --git a/internal/receipt/store.go b/observation/store.go similarity index 80% rename from internal/receipt/store.go rename to observation/store.go index ed09246..c04f353 100644 --- a/internal/receipt/store.go +++ b/observation/store.go @@ -1,4 +1,4 @@ -package receipt +package observation import ( "bytes" @@ -16,36 +16,33 @@ import ( // Store materializes immutable receipt objects and mutable per-run references. type Store struct { - Root string + root string } +// NewStore returns the receipt store rooted under the supplied .yield directory. func NewStore(yieldDir string) *Store { - return &Store{Root: filepath.Join(yieldDir, "receipts")} -} - -func StoreForRunsDir(runsDir string) *Store { - return NewStore(filepath.Dir(runsDir)) + return &Store{root: filepath.Join(yieldDir, "receipts")} } // Materialize projects and durably stores the receipt for one exact prefix. -func (s *Store) Materialize(snapshot Snapshot) (*RunReceipt, []byte, error) { - r, err := Project(snapshot) +func (s *Store) Materialize(journalPrefix []byte) (RunReceipt, []byte, error) { + r, err := Project(journalPrefix) if err != nil { - return nil, nil, err + return RunReceipt{}, nil, err } raw, err := CanonicalBytes(r) if err != nil { - return nil, nil, err + return RunReceipt{}, nil, err } if err := s.Put(r, raw); err != nil { - return nil, nil, err + return RunReceipt{}, nil, err } return r, raw, nil } // Put durably stores an already-sealed receipt. -func (s *Store) Put(r *RunReceipt, raw []byte) error { - if r == nil || r.ReceiptDigest == "" || r.Run.ID == "" { +func (s *Store) Put(r RunReceipt, raw []byte) error { + if r.ReceiptDigest == "" || r.Run.ID == "" { return fmt.Errorf("receipt store: incomplete receipt") } if err := r.Validate(); err != nil { @@ -58,8 +55,8 @@ func (s *Store) Put(r *RunReceipt, raw []byte) error { if len(digest) != 64 { return fmt.Errorf("receipt store: invalid receipt digest") } - objects := filepath.Join(s.Root, "objects", "sha256", digest[:2]) - references := filepath.Join(s.Root, "runs") + objects := filepath.Join(s.root, "objects", "sha256", digest[:2]) + references := filepath.Join(s.root, "runs") if err := secureMkdirAll(objects); err != nil { return err } @@ -74,46 +71,38 @@ func (s *Store) Put(r *RunReceipt, raw []byte) error { } // LoadRun reads the latest materialized receipt for a run. -func (s *Store) LoadRun(runID string) (*RunReceipt, []byte, error) { - ref, err := os.ReadFile(filepath.Join(s.Root, "runs", runID+".ref")) +func (s *Store) LoadRun(runID string) (RunReceipt, []byte, error) { + ref, err := os.ReadFile(filepath.Join(s.root, "runs", runID+".ref")) if err != nil { - return nil, nil, err + return RunReceipt{}, nil, err } digest := strings.TrimSpace(string(ref)) return s.LoadDigest(digest) } // LoadDigest reads and verifies an immutable receipt object. -func (s *Store) LoadDigest(receiptDigest string) (*RunReceipt, []byte, error) { +func (s *Store) LoadDigest(receiptDigest string) (RunReceipt, []byte, error) { if !validDigest(receiptDigest) { - return nil, nil, fmt.Errorf("receipt store: invalid receipt digest") + return RunReceipt{}, nil, fmt.Errorf("receipt store: invalid receipt digest") } hexDigest := strings.TrimPrefix(receiptDigest, "sha256:") - raw, err := os.ReadFile(filepath.Join(s.Root, "objects", "sha256", hexDigest[:2], hexDigest+".json")) + raw, err := os.ReadFile(filepath.Join(s.root, "objects", "sha256", hexDigest[:2], hexDigest+".json")) if err != nil { - return nil, nil, err - } - var r RunReceipt - decoder := json.NewDecoder(bytes.NewReader(raw)) - decoder.DisallowUnknownFields() - if err := decoder.Decode(&r); err != nil { - return nil, nil, fmt.Errorf("receipt store: decode object: %w", err) + return RunReceipt{}, nil, err } - if err := expectEOF(decoder); err != nil { - return nil, nil, err + r, err := Parse(raw) + if err != nil { + return RunReceipt{}, nil, fmt.Errorf("receipt store: decode object: %w", err) } if r.ReceiptDigest != receiptDigest { - return nil, nil, fmt.Errorf("receipt store: reference and object digest differ") + return RunReceipt{}, nil, fmt.Errorf("receipt store: reference and object digest differ") } - if err := VerifyCanonical(&r, raw); err != nil { - return nil, nil, err - } - return &r, raw, nil + return r, raw, nil } // ListRuns returns run IDs with materialized latest-receipt references. func (s *Store) ListRuns() ([]string, error) { - entries, err := os.ReadDir(filepath.Join(s.Root, "runs")) + entries, err := os.ReadDir(filepath.Join(s.root, "runs")) if errors.Is(err, os.ErrNotExist) { return []string{}, nil } diff --git a/internal/receipt/store_test.go b/observation/store_test.go similarity index 86% rename from internal/receipt/store_test.go rename to observation/store_test.go index 4132a4e..c512fff 100644 --- a/internal/receipt/store_test.go +++ b/observation/store_test.go @@ -1,4 +1,4 @@ -package receipt +package observation import ( "bytes" @@ -16,7 +16,7 @@ func TestStoreMaterializationConverges(t *testing.T) { t0 := time.Date(2026, 8, 20, 10, 0, 0, 0, time.UTC) skill := protocol.SkillRef{Name: "store", Digest: protocol.DigestBytes([]byte("store"))} events := []runlog.Event{event(t, 1, runlog.RunStarted, t0, map[string]any{"run_id": "run_store", "skill": skill})} - snapshot := Snapshot{Events: events, Bytes: journalBytes(t, events)} + snapshot := journalBytes(t, events) store := NewStore(t.TempDir()) first, firstBytes, err := store.Materialize(snapshot) if err != nil { @@ -39,8 +39,8 @@ func TestStoreMaterializationConverges(t *testing.T) { if runtime.GOOS != "windows" { digest := first.ReceiptDigest[len("sha256:"):] for _, path := range []string{ - filepath.Join(store.Root, "objects", "sha256", digest[:2], digest+".json"), - filepath.Join(store.Root, "runs", "run_store.ref"), + filepath.Join(store.root, "objects", "sha256", digest[:2], digest+".json"), + filepath.Join(store.root, "runs", "run_store.ref"), } { info, err := os.Stat(path) if err != nil { @@ -57,13 +57,13 @@ func TestStoreRepairsObjectWithoutRunReference(t *testing.T) { t0 := time.Date(2026, 8, 20, 10, 0, 0, 0, time.UTC) skill := protocol.SkillRef{Name: "store", Digest: protocol.DigestBytes([]byte("store"))} events := []runlog.Event{event(t, 1, runlog.RunStarted, t0, map[string]any{"run_id": "run_store", "skill": skill})} - snapshot := Snapshot{Events: events, Bytes: journalBytes(t, events)} + snapshot := journalBytes(t, events) store := NewStore(t.TempDir()) r, _, err := store.Materialize(snapshot) if err != nil { t.Fatal(err) } - ref := filepath.Join(store.Root, "runs", "run_store.ref") + ref := filepath.Join(store.root, "runs", "run_store.ref") if err := os.Remove(ref); err != nil { t.Fatal(err) } @@ -81,12 +81,12 @@ func TestStoreRejectsContentMismatchAtDigestPath(t *testing.T) { skill := protocol.SkillRef{Name: "store", Digest: protocol.DigestBytes([]byte("store"))} events := []runlog.Event{event(t, 1, runlog.RunStarted, t0, map[string]any{"run_id": "run_store", "skill": skill})} store := NewStore(t.TempDir()) - r, raw, err := store.Materialize(Snapshot{Events: events, Bytes: journalBytes(t, events)}) + r, raw, err := store.Materialize(journalBytes(t, events)) if err != nil { t.Fatal(err) } digest := r.ReceiptDigest[len("sha256:"):] - path := filepath.Join(store.Root, "objects", "sha256", digest[:2], digest+".json") + path := filepath.Join(store.root, "objects", "sha256", digest[:2], digest+".json") if err := os.WriteFile(path, []byte("different"), 0o600); err != nil { t.Fatal(err) } @@ -99,7 +99,7 @@ func TestStoreRejectsUnsealedReceiptDigest(t *testing.T) { t0 := time.Date(2026, 8, 20, 10, 0, 0, 0, time.UTC) skill := protocol.SkillRef{Name: "store", Digest: protocol.DigestBytes([]byte("store"))} events := []runlog.Event{event(t, 1, runlog.RunStarted, t0, map[string]any{"run_id": "run_store", "skill": skill})} - r, err := Project(Snapshot{Events: events, Bytes: journalBytes(t, events)}) + r, err := Project(journalBytes(t, events)) if err != nil { t.Fatal(err) } diff --git a/sdk/python/test_observation.py b/sdk/python/test_observation.py new file mode 100644 index 0000000..0b14435 --- /dev/null +++ b/sdk/python/test_observation.py @@ -0,0 +1,44 @@ +import json +import unittest +from pathlib import Path + +from yieldskill import ReceiptError, parse_run_receipt, verify_run_receipt + + +FIXTURE = ( + Path(__file__).parents[2] + / "ir" + / "yield.observation.v1" + / "testdata" + / "run-receipt.canonical.jsonl" +).read_bytes()[:-1] + + +class ReceiptTests(unittest.TestCase): + def test_shared_go_canonical_receipt(self): + receipt = parse_run_receipt(FIXTURE) + self.assertEqual( + receipt["receipt_digest"], + "sha256:a0995d74d51e472f61dc2001f82b93fd8086f9c2f2816a22bb7e5c7fa3c01cdd", + ) + self.assertEqual(receipt["operations"][0]["kind"], "agent_task") + verify_run_receipt(receipt, FIXTURE) + + def test_rejects_unknown_noncanonical_and_digest_mismatch(self): + document = json.loads(FIXTURE) + document["prompt"] = "secret" + with self.assertRaises(ReceiptError): + parse_run_receipt(json.dumps(document, separators=(",", ":")).encode()) + with self.assertRaisesRegex(ReceiptError, "canonical"): + parse_run_receipt(FIXTURE + b"\n") + with self.assertRaisesRegex(ReceiptError, "digest"): + parse_run_receipt( + FIXTURE.replace( + document["receipt_digest"].encode(), + ("sha256:" + "0" * 64).encode(), + ) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/sdk/python/yieldskill/__init__.py b/sdk/python/yieldskill/__init__.py index 3ea4a0d..d5753ab 100644 --- a/sdk/python/yieldskill/__init__.py +++ b/sdk/python/yieldskill/__init__.py @@ -20,12 +20,18 @@ from dataclasses import dataclass from typing import Any, Callable, Optional +from .observation import ReceiptError, RunReceipt, parse_run_receipt, verify_run_receipt + __all__ = [ "Blocked", "Refused", "CommandResult", "Context", + "ReceiptError", + "RunReceipt", "define_skill", + "parse_run_receipt", + "verify_run_receipt", ] _PROTOCOL = "yield.v1" diff --git a/sdk/python/yieldskill/observation.py b/sdk/python/yieldskill/observation.py new file mode 100644 index 0000000..e17f86e --- /dev/null +++ b/sdk/python/yieldskill/observation.py @@ -0,0 +1,625 @@ +"""Strict yield.observation.v1 receipt types, parsing, and verification.""" + +from __future__ import annotations + +import hashlib +import json +import re +from datetime import datetime +from typing import Any, Literal, Mapping, TypedDict, cast + + +OperationKind = Literal["ask_user", "agent_task", "run_command"] +LifecyclePhase = Literal[ + "initializing", + "initialization_failed", + "awaiting_response", + "advancing", + "recoverable_error", + "diverged", + "terminal", +] +TerminalDisposition = Literal["completed", "blocked", "refused"] +TerminalCause = Literal[ + "completed", "blocked", "refused", "requirement_failed", "completion_unproven" +] +FailureCode = Literal[ + "manifest_invalid", + "manifest_read_failed", + "runtime_version_missing", + "runtime_incompatible", + "runner_missing", + "source_digest_failed", + "initialization_failed", + "invalid_program_output", + "execution_timeout", + "subprocess_failed", + "execution_failed", + "command_execution_failed", +] +RequirementResult = Literal["passed", "failed"] +ResponseRejectionReason = Literal[ + "wrong-run", + "stale-response", + "duplicate-response", + "wrong-request", + "schema-invalid", + "digest-mismatch", + "completion-unproven", + "run-closed", + "no-pending-operation", +] +ExperimentRole = Literal["baseline", "candidate"] + + +class JournalBinding(TypedDict): + run_id: str + head_sequence: int + head_digest: str + + +class _RunIdentityRequired(TypedDict): + id: str + + +class RunIdentity(_RunIdentityRequired, total=False): + input_digest: str + + +class ProfileDigest(TypedDict): + profile: str + value: str + + +class _SkillIdentityRequired(TypedDict): + name: str + + +class SkillIdentity(_SkillIdentityRequired, total=False): + version: str + binding_digest: str + source_digest: ProfileDigest + + +class RuntimeIdentity(TypedDict, total=False): + supervisor_version: str + required_version: str + compatible: bool + + +class _TimingSummaryRequired(TypedDict): + started_at: str + last_observed_at: str + + +class TimingSummary(_TimingSummaryRequired, total=False): + ended_at: str + elapsed_ms: int + clock_anomaly: bool + + +class _OperationObservationRequired(TypedDict): + sequence: int + kind: OperationKind + operation_key_digest: str + requested_at: str + + +class OperationObservation(_OperationObservationRequired, total=False): + completed_at: str + elapsed_ms: int + result_digest: str + clock_anomaly: bool + + +class OperationSummary(TypedDict): + kind: OperationKind + requested: int + completed: int + total_elapsed_ms: int + + +class _OutcomeSummaryRequired(TypedDict): + phase: LifecyclePhase + + +class OutcomeSummary(_OutcomeSummaryRequired, total=False): + terminal_disposition: TerminalDisposition + terminal_cause: TerminalCause + result_digest: str + failure_code: FailureCode + + +class _RequirementOutcomeRequired(TypedDict): + outcome: RequirementResult + claim_digest: str + + +class RequirementOutcome(_RequirementOutcomeRequired, total=False): + evidence_digest: str + + +class ResponseRejectionSummary(TypedDict): + reason: ResponseRejectionReason + count: int + + +class DivergenceOutcome(TypedDict): + sequence: int + expected_digest: str + got_digest: str + + +class _ExperimentContextRequired(TypedDict): + experiment_id: str + variant_id: str + role: ExperimentRole + + +class ExperimentContext(_ExperimentContextRequired, total=False): + cohort_id: str + baseline_variant_id: str + parent_skill_version: str + + +class _RunReceiptRequired(TypedDict): + schema: Literal["yield.observation.v1"] + kind: Literal["run_receipt"] + receipt_digest: str + journal: JournalBinding + run: RunIdentity + skill: SkillIdentity + timing: TimingSummary + operations: list[OperationObservation] + operation_summaries: list[OperationSummary] + outcome: OutcomeSummary + requirements: list[RequirementOutcome] + response_rejections: list[ResponseRejectionSummary] + divergences: list[DivergenceOutcome] + + +class RunReceipt(_RunReceiptRequired, total=False): + runtime: RuntimeIdentity + experiment: ExperimentContext + + +class ReceiptError(ValueError): + """The receipt is not strict, canonical, or digest-valid.""" + + +_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") +_SEMVER = re.compile(r"^\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.-]+)?$") +_SOURCE_PROFILE = re.compile(r"^yield\.skill-source\.v[0-9]+$") +_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") +_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,9})?(?:Z|[+-]\d{2}:\d{2})$") + +_OPERATION_KINDS = {"ask_user", "agent_task", "run_command"} +_PHASES = { + "initializing", + "initialization_failed", + "awaiting_response", + "advancing", + "recoverable_error", + "diverged", + "terminal", +} +_DISPOSITIONS = {"completed", "blocked", "refused"} +_TERMINAL_CAUSES = { + "completed", + "blocked", + "refused", + "requirement_failed", + "completion_unproven", +} +_FAILURE_CODES = { + "manifest_invalid", + "manifest_read_failed", + "runtime_version_missing", + "runtime_incompatible", + "runner_missing", + "source_digest_failed", + "initialization_failed", + "invalid_program_output", + "execution_timeout", + "subprocess_failed", + "execution_failed", + "command_execution_failed", +} +_REJECTION_REASONS = { + "wrong-run", + "stale-response", + "duplicate-response", + "wrong-request", + "schema-invalid", + "digest-mismatch", + "completion-unproven", + "run-closed", + "no-pending-operation", +} + + +def _object(value: Any, path: str, allowed: set[str], required: set[str]) -> dict[str, Any]: + if not isinstance(value, dict): + raise ReceiptError(f"{path}: must be an object") + unknown = set(value) - allowed + if unknown: + raise ReceiptError(f"{path}: unknown field {sorted(unknown)[0]!r}") + missing = required - set(value) + if missing: + raise ReceiptError(f"{path}: missing field {sorted(missing)[0]!r}") + return value + + +def _string(value: Any, path: str, pattern: re.Pattern[str] | None = None) -> str: + if ( + not isinstance(value, str) + or not value + or (pattern and not pattern.fullmatch(value)) + or any(0xD800 <= ord(character) <= 0xDFFF for character in value) + ): + raise ReceiptError(f"{path}: must be a valid non-empty string") + return value + + +def _timestamp(value: Any, path: str) -> str: + text = _string(value, path, _TIMESTAMP) + try: + datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError as error: + raise ReceiptError(f"{path}: must be an RFC 3339 timestamp") from error + return text + + +def _integer(value: Any, path: str, minimum: int) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < minimum: + raise ReceiptError(f"{path}: must be an integer >= {minimum}") + return value + + +def _boolean(value: Any, path: str) -> bool: + if not isinstance(value, bool): + raise ReceiptError(f"{path}: must be a boolean") + return value + + +def _one_of(value: Any, path: str, values: set[str]) -> str: + if not isinstance(value, str) or value not in values: + raise ReceiptError(f"{path}: invalid value") + return value + + +def _optional(value: dict[str, Any], key: str, validator: Any, path: str) -> None: + if key in value: + validator(value[key], f"{path}.{key}") + + +def _array(value: dict[str, Any], key: str, validator: Any, path: str = "$") -> None: + items = value[key] + if not isinstance(items, list): + raise ReceiptError(f"{path}.{key}: must be an array") + for index, item in enumerate(items): + validator(item, f"{path}.{key}[{index}]") + + +def _validate_receipt(value: Any) -> None: + receipt = _object( + value, + "$", + { + "schema", + "kind", + "receipt_digest", + "journal", + "run", + "skill", + "runtime", + "timing", + "operations", + "operation_summaries", + "outcome", + "requirements", + "response_rejections", + "divergences", + "experiment", + }, + { + "schema", + "kind", + "receipt_digest", + "journal", + "run", + "skill", + "timing", + "operations", + "operation_summaries", + "outcome", + "requirements", + "response_rejections", + "divergences", + }, + ) + if receipt["schema"] != "yield.observation.v1" or receipt["kind"] != "run_receipt": + raise ReceiptError("$: invalid schema or kind") + _string(receipt["receipt_digest"], "$.receipt_digest", _DIGEST) + + journal = _object( + receipt["journal"], + "$.journal", + {"run_id", "head_sequence", "head_digest"}, + {"run_id", "head_sequence", "head_digest"}, + ) + run_id = _string(journal["run_id"], "$.journal.run_id") + _integer(journal["head_sequence"], "$.journal.head_sequence", 1) + _string(journal["head_digest"], "$.journal.head_digest", _DIGEST) + + run = _object(receipt["run"], "$.run", {"id", "input_digest"}, {"id"}) + if _string(run["id"], "$.run.id") != run_id: + raise ReceiptError("$.run.id: must match journal.run_id") + _optional(run, "input_digest", lambda v, p: _string(v, p, _DIGEST), "$.run") + + skill = _object( + receipt["skill"], + "$.skill", + {"name", "version", "binding_digest", "source_digest"}, + {"name"}, + ) + _string(skill["name"], "$.skill.name") + _optional(skill, "version", lambda v, p: _string(v, p, _SEMVER), "$.skill") + _optional(skill, "binding_digest", lambda v, p: _string(v, p, _DIGEST), "$.skill") + if "source_digest" in skill: + source = _object( + skill["source_digest"], + "$.skill.source_digest", + {"profile", "value"}, + {"profile", "value"}, + ) + _string(source["profile"], "$.skill.source_digest.profile", _SOURCE_PROFILE) + _string(source["value"], "$.skill.source_digest.value", _DIGEST) + + if "runtime" in receipt: + runtime = _object( + receipt["runtime"], + "$.runtime", + {"supervisor_version", "required_version", "compatible"}, + set(), + ) + if not runtime: + raise ReceiptError("$.runtime: must not be empty") + _optional( + runtime, + "supervisor_version", + lambda v, p: _string(v, p, _SEMVER), + "$.runtime", + ) + _optional( + runtime, + "required_version", + lambda v, p: _string(v, p, _SEMVER), + "$.runtime", + ) + _optional(runtime, "compatible", _boolean, "$.runtime") + if "compatible" in runtime and not { + "supervisor_version", + "required_version", + }.issubset(runtime): + raise ReceiptError("$.runtime.compatible: requires both runtime versions") + + timing = _object( + receipt["timing"], + "$.timing", + {"started_at", "last_observed_at", "ended_at", "elapsed_ms", "clock_anomaly"}, + {"started_at", "last_observed_at"}, + ) + _timestamp(timing["started_at"], "$.timing.started_at") + _timestamp(timing["last_observed_at"], "$.timing.last_observed_at") + _optional(timing, "ended_at", _timestamp, "$.timing") + _optional(timing, "elapsed_ms", lambda v, p: _integer(v, p, 0), "$.timing") + _optional(timing, "clock_anomaly", _boolean, "$.timing") + if timing.get("clock_anomaly") is False: + raise ReceiptError("$.timing.clock_anomaly: false must be omitted") + + def operation(item: Any, path: str) -> None: + observed = _object( + item, + path, + { + "sequence", + "kind", + "operation_key_digest", + "requested_at", + "completed_at", + "elapsed_ms", + "result_digest", + "clock_anomaly", + }, + {"sequence", "kind", "operation_key_digest", "requested_at"}, + ) + _integer(observed["sequence"], f"{path}.sequence", 1) + _one_of(observed["kind"], f"{path}.kind", _OPERATION_KINDS) + _string(observed["operation_key_digest"], f"{path}.operation_key_digest", _DIGEST) + _timestamp(observed["requested_at"], f"{path}.requested_at") + _optional(observed, "completed_at", _timestamp, path) + _optional(observed, "elapsed_ms", lambda v, p: _integer(v, p, 0), path) + _optional(observed, "result_digest", lambda v, p: _string(v, p, _DIGEST), path) + _optional(observed, "clock_anomaly", _boolean, path) + if observed.get("clock_anomaly") is False: + raise ReceiptError(f"{path}.clock_anomaly: false must be omitted") + + _array(receipt, "operations", operation) + + def operation_summary(item: Any, path: str) -> None: + summary = _object( + item, + path, + {"kind", "requested", "completed", "total_elapsed_ms"}, + {"kind", "requested", "completed", "total_elapsed_ms"}, + ) + _one_of(summary["kind"], f"{path}.kind", _OPERATION_KINDS) + requested = _integer(summary["requested"], f"{path}.requested", 0) + completed = _integer(summary["completed"], f"{path}.completed", 0) + if completed > requested: + raise ReceiptError(f"{path}.completed: must not exceed requested") + _integer(summary["total_elapsed_ms"], f"{path}.total_elapsed_ms", 0) + + _array(receipt, "operation_summaries", operation_summary) + + outcome = _object( + receipt["outcome"], + "$.outcome", + { + "phase", + "terminal_disposition", + "terminal_cause", + "result_digest", + "failure_code", + }, + {"phase"}, + ) + phase = _one_of(outcome["phase"], "$.outcome.phase", _PHASES) + _optional( + outcome, + "terminal_disposition", + lambda v, p: _one_of(v, p, _DISPOSITIONS), + "$.outcome", + ) + _optional( + outcome, + "terminal_cause", + lambda v, p: _one_of(v, p, _TERMINAL_CAUSES), + "$.outcome", + ) + _optional(outcome, "result_digest", lambda v, p: _string(v, p, _DIGEST), "$.outcome") + _optional( + outcome, + "failure_code", + lambda v, p: _one_of(v, p, _FAILURE_CODES), + "$.outcome", + ) + if phase == "terminal": + if ( + not {"terminal_disposition", "terminal_cause"}.issubset(outcome) + or "ended_at" not in timing + ): + raise ReceiptError( + "$.outcome: terminal outcome requires disposition, cause, and end time" + ) + elif "terminal_disposition" in outcome or "terminal_cause" in outcome: + raise ReceiptError("$.outcome: nonterminal outcome cannot contain terminal fields") + if phase in {"initialization_failed", "recoverable_error"} and "failure_code" not in outcome: + raise ReceiptError("$.outcome.failure_code: required for failure phase") + + def requirement(item: Any, path: str) -> None: + result = _object( + item, + path, + {"outcome", "claim_digest", "evidence_digest"}, + {"outcome", "claim_digest"}, + ) + _one_of(result["outcome"], f"{path}.outcome", {"passed", "failed"}) + _string(result["claim_digest"], f"{path}.claim_digest", _DIGEST) + _optional(result, "evidence_digest", lambda v, p: _string(v, p, _DIGEST), path) + + _array(receipt, "requirements", requirement) + + def rejection(item: Any, path: str) -> None: + result = _object(item, path, {"reason", "count"}, {"reason", "count"}) + _one_of(result["reason"], f"{path}.reason", _REJECTION_REASONS) + _integer(result["count"], f"{path}.count", 1) + + _array(receipt, "response_rejections", rejection) + + def divergence(item: Any, path: str) -> None: + result = _object( + item, + path, + {"sequence", "expected_digest", "got_digest"}, + {"sequence", "expected_digest", "got_digest"}, + ) + _integer(result["sequence"], f"{path}.sequence", 1) + _string(result["expected_digest"], f"{path}.expected_digest", _DIGEST) + _string(result["got_digest"], f"{path}.got_digest", _DIGEST) + + _array(receipt, "divergences", divergence) + + if "experiment" in receipt: + experiment = _object( + receipt["experiment"], + "$.experiment", + { + "experiment_id", + "cohort_id", + "variant_id", + "role", + "baseline_variant_id", + "parent_skill_version", + }, + {"experiment_id", "variant_id", "role"}, + ) + _string(experiment["experiment_id"], "$.experiment.experiment_id", _IDENTIFIER) + _optional( + experiment, + "cohort_id", + lambda v, p: _string(v, p, _IDENTIFIER), + "$.experiment", + ) + _string(experiment["variant_id"], "$.experiment.variant_id", _IDENTIFIER) + _one_of(experiment["role"], "$.experiment.role", {"baseline", "candidate"}) + _optional( + experiment, + "baseline_variant_id", + lambda v, p: _string(v, p, _IDENTIFIER), + "$.experiment", + ) + _optional( + experiment, + "parent_skill_version", + lambda v, p: _string(v, p, _SEMVER), + "$.experiment", + ) + + +def _canonical(value: Mapping[str, Any]) -> bytes: + try: + return json.dumps( + value, + ensure_ascii=False, + allow_nan=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + except (TypeError, ValueError) as error: + raise ReceiptError(f"receipt contains a non-JSON value: {error}") from error + + +def _object_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + value: dict[str, Any] = {} + for key, item in pairs: + if key in value: + raise ReceiptError(f"duplicate field {key!r}") + value[key] = item + return value + + +def verify_run_receipt(receipt: RunReceipt, canonical: bytes) -> None: + """Verify a typed receipt against its exact canonical bytes and digest.""" + _validate_receipt(receipt) + encoded = _canonical(receipt) + if encoded != canonical: + raise ReceiptError("receipt bytes are not canonical JSON") + body = dict(receipt) + expected = cast(str, body.pop("receipt_digest")) + actual = "sha256:" + hashlib.sha256(_canonical(body)).hexdigest() + if actual != expected: + raise ReceiptError("receipt digest verification failed") + + +def parse_run_receipt(canonical: bytes) -> RunReceipt: + """Parse and verify one canonical yield.observation.v1 receipt.""" + try: + document = json.loads(canonical.decode("utf-8"), object_pairs_hook=_object_pairs) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise ReceiptError(f"receipt does not decode: {error}") from error + _validate_receipt(document) + receipt = cast(RunReceipt, document) + verify_run_receipt(receipt, canonical) + return receipt diff --git a/sdk/rust/src/lib.rs b/sdk/rust/src/lib.rs index 50f4510..ac9b439 100644 --- a/sdk/rust/src/lib.rs +++ b/sdk/rust/src/lib.rs @@ -18,6 +18,9 @@ use sha2::{Digest, Sha256}; use std::io::Write; use std::process::exit; +pub mod observation; +pub use observation::{ReceiptError, RunReceipt}; + pub const PROTOCOL: &str = "yield.v1"; fn verify_supervisor_identity() { diff --git a/sdk/rust/src/observation.rs b/sdk/rust/src/observation.rs new file mode 100644 index 0000000..c8f1220 --- /dev/null +++ b/sdk/rust/src/observation.rs @@ -0,0 +1,668 @@ +//! Strict types and verification for `yield.observation.v1` run receipts. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use std::fmt::{Display, Formatter}; + +#[derive(Debug)] +pub struct ReceiptError(String); + +impl ReceiptError { + fn new(message: impl Into) -> Self { + Self(message.into()) + } +} + +impl Display for ReceiptError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + formatter.write_str(&self.0) + } +} + +impl std::error::Error for ReceiptError {} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OperationKind { + AskUser, + AgentTask, + RunCommand, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LifecyclePhase { + Initializing, + InitializationFailed, + AwaitingResponse, + Advancing, + RecoverableError, + Diverged, + Terminal, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TerminalDisposition { + Completed, + Blocked, + Refused, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TerminalCause { + Completed, + Blocked, + Refused, + RequirementFailed, + CompletionUnproven, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum FailureCode { + ManifestInvalid, + ManifestReadFailed, + RuntimeVersionMissing, + RuntimeIncompatible, + RunnerMissing, + SourceDigestFailed, + InitializationFailed, + InvalidProgramOutput, + ExecutionTimeout, + SubprocessFailed, + ExecutionFailed, + CommandExecutionFailed, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RequirementResult { + Passed, + Failed, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum ResponseRejectionReason { + WrongRun, + StaleResponse, + DuplicateResponse, + WrongRequest, + SchemaInvalid, + DigestMismatch, + CompletionUnproven, + RunClosed, + NoPendingOperation, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ExperimentRole { + Baseline, + Candidate, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct JournalBinding { + pub run_id: String, + pub head_sequence: u64, + pub head_digest: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RunIdentity { + pub id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub input_digest: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ProfileDigest { + pub profile: String, + pub value: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SkillIdentity { + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub binding_digest: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub source_digest: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RuntimeIdentity { + #[serde(skip_serializing_if = "Option::is_none")] + pub supervisor_version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub required_version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub compatible: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TimingSummary { + pub started_at: String, + pub last_observed_at: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub ended_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub elapsed_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub clock_anomaly: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OperationObservation { + pub sequence: u64, + pub kind: OperationKind, + pub operation_key_digest: String, + pub requested_at: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub completed_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub elapsed_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub result_digest: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub clock_anomaly: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OperationSummary { + pub kind: OperationKind, + pub requested: u64, + pub completed: u64, + pub total_elapsed_ms: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OutcomeSummary { + pub phase: LifecyclePhase, + #[serde(skip_serializing_if = "Option::is_none")] + pub terminal_disposition: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub terminal_cause: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub result_digest: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub failure_code: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RequirementOutcome { + pub outcome: RequirementResult, + pub claim_digest: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub evidence_digest: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ResponseRejectionSummary { + pub reason: ResponseRejectionReason, + pub count: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DivergenceOutcome { + pub sequence: u64, + pub expected_digest: String, + pub got_digest: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ExperimentContext { + pub experiment_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub cohort_id: Option, + pub variant_id: String, + pub role: ExperimentRole, + #[serde(skip_serializing_if = "Option::is_none")] + pub baseline_variant_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub parent_skill_version: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RunReceipt { + pub schema: String, + pub kind: String, + pub receipt_digest: String, + pub journal: JournalBinding, + pub run: RunIdentity, + pub skill: SkillIdentity, + #[serde(skip_serializing_if = "Option::is_none")] + pub runtime: Option, + pub timing: TimingSummary, + pub operations: Vec, + pub operation_summaries: Vec, + pub outcome: OutcomeSummary, + pub requirements: Vec, + pub response_rejections: Vec, + pub divergences: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub experiment: Option, +} + +impl RunReceipt { + /// Parse and verify one canonical `yield.observation.v1` receipt. + pub fn parse_and_verify(bytes: &[u8]) -> Result { + let receipt: Self = serde_json::from_slice(bytes) + .map_err(|error| ReceiptError::new(format!("receipt does not decode: {error}")))?; + receipt.verify(bytes)?; + Ok(receipt) + } + + /// Verify this typed receipt against its exact canonical bytes and digest. + pub fn verify(&self, bytes: &[u8]) -> Result<(), ReceiptError> { + self.validate()?; + let value = serde_json::to_value(self) + .map_err(|error| ReceiptError::new(format!("receipt does not encode: {error}")))?; + let canonical = canonical_bytes(&value)?; + if canonical != bytes { + return Err(ReceiptError::new("receipt bytes are not canonical JSON")); + } + let mut body = value; + let Value::Object(ref mut object) = body else { + return Err(ReceiptError::new("receipt must be an object")); + }; + object.remove("receipt_digest"); + let actual = format!( + "sha256:{}", + hex::encode(Sha256::digest(canonical_bytes(&body)?)) + ); + if actual != self.receipt_digest { + return Err(ReceiptError::new("receipt digest verification failed")); + } + Ok(()) + } + + fn validate(&self) -> Result<(), ReceiptError> { + if self.schema != "yield.observation.v1" || self.kind != "run_receipt" { + return Err(ReceiptError::new("invalid receipt schema or kind")); + } + require_digest(&self.receipt_digest, "receipt_digest")?; + if self.journal.run_id.is_empty() || self.journal.head_sequence == 0 { + return Err(ReceiptError::new("incomplete journal binding")); + } + require_digest(&self.journal.head_digest, "journal.head_digest")?; + if self.run.id != self.journal.run_id || self.skill.name.is_empty() { + return Err(ReceiptError::new("incomplete run or skill identity")); + } + optional_digest(self.run.input_digest.as_deref(), "run.input_digest")?; + optional_digest(self.skill.binding_digest.as_deref(), "skill.binding_digest")?; + if let Some(version) = &self.skill.version { + require_semver(version, "skill.version")?; + } + if let Some(source) = &self.skill.source_digest { + if !source_profile(&source.profile) { + return Err(ReceiptError::new("invalid skill.source_digest.profile")); + } + require_digest(&source.value, "skill.source_digest.value")?; + } + if let Some(runtime) = &self.runtime { + if runtime.supervisor_version.is_none() + && runtime.required_version.is_none() + && runtime.compatible.is_none() + { + return Err(ReceiptError::new("runtime identity must not be empty")); + } + if let Some(version) = &runtime.supervisor_version { + require_semver(version, "runtime.supervisor_version")?; + } + if let Some(version) = &runtime.required_version { + require_semver(version, "runtime.required_version")?; + } + if runtime.compatible.is_some() + && (runtime.supervisor_version.is_none() || runtime.required_version.is_none()) + { + return Err(ReceiptError::new( + "runtime.compatible requires both versions", + )); + } + } + require_timestamp(&self.timing.started_at, "timing.started_at")?; + require_timestamp(&self.timing.last_observed_at, "timing.last_observed_at")?; + if let Some(value) = &self.timing.ended_at { + require_timestamp(value, "timing.ended_at")?; + } + if self.timing.clock_anomaly == Some(false) { + return Err(ReceiptError::new( + "timing.clock_anomaly false must be omitted", + )); + } + for (index, operation) in self.operations.iter().enumerate() { + if operation.sequence == 0 { + return Err(ReceiptError::new(format!( + "operations[{index}].sequence must be positive" + ))); + } + require_digest( + &operation.operation_key_digest, + "operation.operation_key_digest", + )?; + require_timestamp(&operation.requested_at, "operation.requested_at")?; + if let Some(value) = &operation.completed_at { + require_timestamp(value, "operation.completed_at")?; + } + optional_digest( + operation.result_digest.as_deref(), + "operation.result_digest", + )?; + if operation.clock_anomaly == Some(false) { + return Err(ReceiptError::new( + "operation.clock_anomaly false must be omitted", + )); + } + } + for summary in &self.operation_summaries { + if summary.completed > summary.requested { + return Err(ReceiptError::new( + "operation summary completed exceeds requested", + )); + } + } + if self.outcome.phase == LifecyclePhase::Terminal { + if self.outcome.terminal_disposition.is_none() + || self.outcome.terminal_cause.is_none() + || self.timing.ended_at.is_none() + { + return Err(ReceiptError::new("terminal outcome is incomplete")); + } + } else if self.outcome.terminal_disposition.is_some() + || self.outcome.terminal_cause.is_some() + { + return Err(ReceiptError::new( + "nonterminal outcome contains terminal fields", + )); + } + if matches!( + self.outcome.phase, + LifecyclePhase::InitializationFailed | LifecyclePhase::RecoverableError + ) && self.outcome.failure_code.is_none() + { + return Err(ReceiptError::new("failure phase has no failure_code")); + } + optional_digest( + self.outcome.result_digest.as_deref(), + "outcome.result_digest", + )?; + for requirement in &self.requirements { + require_digest(&requirement.claim_digest, "requirement.claim_digest")?; + optional_digest( + requirement.evidence_digest.as_deref(), + "requirement.evidence_digest", + )?; + } + for rejection in &self.response_rejections { + if rejection.count == 0 { + return Err(ReceiptError::new( + "response rejection count must be positive", + )); + } + } + for divergence in &self.divergences { + if divergence.sequence == 0 { + return Err(ReceiptError::new("divergence sequence must be positive")); + } + require_digest(&divergence.expected_digest, "divergence.expected_digest")?; + require_digest(&divergence.got_digest, "divergence.got_digest")?; + } + if let Some(experiment) = &self.experiment { + require_identifier(&experiment.experiment_id, "experiment.experiment_id")?; + require_identifier(&experiment.variant_id, "experiment.variant_id")?; + if let Some(value) = &experiment.cohort_id { + require_identifier(value, "experiment.cohort_id")?; + } + if let Some(value) = &experiment.baseline_variant_id { + require_identifier(value, "experiment.baseline_variant_id")?; + } + if let Some(value) = &experiment.parent_skill_version { + require_semver(value, "experiment.parent_skill_version")?; + } + } + Ok(()) + } +} + +fn require_digest(value: &str, field: &str) -> Result<(), ReceiptError> { + let valid = value.len() == 71 + && value.starts_with("sha256:") + && value[7..] + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)); + if valid { + Ok(()) + } else { + Err(ReceiptError::new(format!("invalid {field}"))) + } +} + +fn optional_digest(value: Option<&str>, field: &str) -> Result<(), ReceiptError> { + match value { + Some(value) => require_digest(value, field), + None => Ok(()), + } +} + +fn require_semver(value: &str, field: &str) -> Result<(), ReceiptError> { + let split = value.find(['-', '+']).unwrap_or(value.len()); + let core = &value[..split]; + let suffix = &value[split..]; + let valid_core = core.split('.').count() == 3 + && core + .split('.') + .all(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_digit())); + let valid_suffix = suffix.is_empty() + || suffix.len() > 1 + && suffix[1..] + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-')); + if valid_core && valid_suffix { + Ok(()) + } else { + Err(ReceiptError::new(format!("invalid {field}"))) + } +} + +fn source_profile(value: &str) -> bool { + value + .strip_prefix("yield.skill-source.v") + .is_some_and(|version| { + !version.is_empty() && version.bytes().all(|byte| byte.is_ascii_digit()) + }) +} + +fn require_identifier(value: &str, field: &str) -> Result<(), ReceiptError> { + let valid = !value.is_empty() + && value.len() <= 128 + && value.as_bytes()[0].is_ascii_alphanumeric() + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b':' | b'-')); + if valid { + Ok(()) + } else { + Err(ReceiptError::new(format!("invalid {field}"))) + } +} + +fn require_timestamp(value: &str, field: &str) -> Result<(), ReceiptError> { + let bytes = value.as_bytes(); + let fixed = bytes.len() >= 20 + && bytes.get(4) == Some(&b'-') + && bytes.get(7) == Some(&b'-') + && bytes.get(10) == Some(&b'T') + && bytes.get(13) == Some(&b':') + && bytes.get(16) == Some(&b':'); + let zone_start = if value.ends_with('Z') { + value.len() - 1 + } else if value.len() >= 6 + && matches!(bytes[value.len() - 6], b'+' | b'-') + && bytes[value.len() - 3] == b':' + { + value.len() - 6 + } else { + 0 + }; + if !fixed || zone_start < 19 { + return Err(ReceiptError::new(format!("invalid {field}"))); + } + let parse = |start: usize, end: usize| { + value[start..end] + .parse::() + .map_err(|_| ReceiptError::new(format!("invalid {field}"))) + }; + let year = parse(0, 4)?; + let month = parse(5, 7)?; + let day = parse(8, 10)?; + let hour = parse(11, 13)?; + let minute = parse(14, 16)?; + let second = parse(17, 19)?; + let leap = year % 4 == 0 && (year % 100 != 0 || year % 400 == 0); + let days = match month { + 1 | 3 | 5 | 7 | 8 | 10 | 12 => 31, + 4 | 6 | 9 | 11 => 30, + 2 if leap => 29, + 2 => 28, + _ => 0, + }; + if day == 0 || day > days || hour > 23 || minute > 59 || second > 59 { + return Err(ReceiptError::new(format!("invalid {field}"))); + } + if !value.ends_with('Z') { + let offset_hour = parse(value.len() - 5, value.len() - 3)?; + let offset_minute = parse(value.len() - 2, value.len())?; + if offset_hour > 23 || offset_minute > 59 { + return Err(ReceiptError::new(format!("invalid {field}"))); + } + } + let fraction = &value[19..zone_start]; + let valid_fraction = fraction.is_empty() + || fraction.starts_with('.') + && fraction.len() <= 10 + && fraction[1..].bytes().all(|byte| byte.is_ascii_digit()); + if valid_fraction { + Ok(()) + } else { + Err(ReceiptError::new(format!("invalid {field}"))) + } +} + +fn canonical_bytes(value: &Value) -> Result, ReceiptError> { + let mut output = String::new(); + write_canonical(&mut output, value)?; + Ok(output.into_bytes()) +} + +fn write_canonical(output: &mut String, value: &Value) -> Result<(), ReceiptError> { + match value { + Value::Null => output.push_str("null"), + Value::Bool(value) => output.push_str(if *value { "true" } else { "false" }), + Value::String(value) => output.push_str( + &serde_json::to_string(value) + .map_err(|error| ReceiptError::new(format!("string does not encode: {error}")))?, + ), + Value::Number(value) => { + if !value.is_i64() && !value.is_u64() { + return Err(ReceiptError::new("receipt contains a non-integer number")); + } + output.push_str(&value.to_string()); + } + Value::Array(values) => { + output.push('['); + for (index, value) in values.iter().enumerate() { + if index > 0 { + output.push(','); + } + write_canonical(output, value)?; + } + output.push(']'); + } + Value::Object(values) => { + let mut keys = values.keys().collect::>(); + keys.sort_by_key(|key| key.encode_utf16().collect::>()); + output.push('{'); + for (index, key) in keys.iter().enumerate() { + if index > 0 { + output.push(','); + } + output.push_str( + &serde_json::to_string(key).map_err(|error| { + ReceiptError::new(format!("key does not encode: {error}")) + })?, + ); + output.push(':'); + write_canonical(output, &values[*key])?; + } + output.push('}'); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + const FIXTURE_FILE: &[u8] = + include_bytes!("../../../ir/yield.observation.v1/testdata/run-receipt.canonical.jsonl"); + + fn fixture() -> &'static [u8] { + FIXTURE_FILE + .strip_suffix(b"\n") + .expect("fixture file newline") + } + + #[test] + fn parses_and_verifies_shared_go_canonical_receipt() { + let receipt = RunReceipt::parse_and_verify(fixture()).expect("valid receipt"); + assert_eq!( + receipt.receipt_digest, + "sha256:a0995d74d51e472f61dc2001f82b93fd8086f9c2f2816a22bb7e5c7fa3c01cdd" + ); + assert_eq!(receipt.operations[0].kind, OperationKind::AgentTask); + receipt.verify(fixture()).expect("verified receipt"); + } + + #[test] + fn rejects_unknown_noncanonical_and_digest_mismatch() { + let mut unknown = fixture()[..fixture().len() - 1].to_vec(); + unknown.extend_from_slice(br#","prompt":"secret"}"#); + assert!(RunReceipt::parse_and_verify(&unknown).is_err()); + + let mut noncanonical = fixture().to_vec(); + noncanonical.push(b'\n'); + assert!(RunReceipt::parse_and_verify(&noncanonical).is_err()); + + let current = b"a0995d74d51e472f61dc2001f82b93fd8086f9c2f2816a22bb7e5c7fa3c01cdd"; + let position = fixture() + .windows(current.len()) + .position(|window| window == current) + .expect("digest"); + let mut mismatch = fixture().to_vec(); + mismatch[position..position + current.len()].fill(b'0'); + let error = RunReceipt::parse_and_verify(&mismatch).expect_err("digest mismatch"); + assert!(error.to_string().contains("digest")); + } +} diff --git a/sdk/typescript/scripts/build.mjs b/sdk/typescript/scripts/build.mjs index 25c89ad..0998df9 100644 --- a/sdk/typescript/scripts/build.mjs +++ b/sdk/typescript/scripts/build.mjs @@ -5,15 +5,17 @@ import { fileURLToPath } from "node:url" const here = dirname(fileURLToPath(import.meta.url)) const root = resolve(here, "..") -const sourcePath = resolve(root, "src/index.ts") const distPath = resolve(root, "dist") -const outputPath = resolve(distPath, "index.js") -const source = readFileSync(sourcePath, "utf8") -const runtime = stripTypeScriptTypes(source, { mode: "transform" }) rmSync(distPath, { recursive: true, force: true }) -mkdirSync(dirname(outputPath), { recursive: true }) -writeFileSync( - outputPath, - "// Generated from src/index.ts by scripts/build.mjs. Do not edit.\n" + runtime, -) +for (const name of ["index", "observation"]) { + const sourcePath = resolve(root, `src/${name}.ts`) + const outputPath = resolve(distPath, `${name}.js`) + const source = readFileSync(sourcePath, "utf8") + const runtime = stripTypeScriptTypes(source, { mode: "transform" }) + mkdirSync(dirname(outputPath), { recursive: true }) + writeFileSync( + outputPath, + `// Generated from src/${name}.ts by scripts/build.mjs. Do not edit.\n` + runtime, + ) +} diff --git a/sdk/typescript/src/index.ts b/sdk/typescript/src/index.ts index 63c4c58..a53b32f 100644 --- a/sdk/typescript/src/index.ts +++ b/sdk/typescript/src/index.ts @@ -16,6 +16,8 @@ import { createHash } from "node:crypto"; import { readFileSync } from "node:fs"; import { exit, env, stdout, stderr } from "node:process"; +export * from "./observation.js"; + export type OpKind = "ask_user" | "agent_task" | "run_command"; export interface SkillRef { diff --git a/sdk/typescript/src/observation.js b/sdk/typescript/src/observation.js new file mode 100644 index 0000000..d23d8ba --- /dev/null +++ b/sdk/typescript/src/observation.js @@ -0,0 +1,3 @@ +// Node executes the TypeScript source directly during local skill runs. The +// package build replaces this bridge with dist/observation.js. +export * from "./observation.ts" diff --git a/sdk/typescript/src/observation.test.mjs b/sdk/typescript/src/observation.test.mjs new file mode 100644 index 0000000..59915bb --- /dev/null +++ b/sdk/typescript/src/observation.test.mjs @@ -0,0 +1,36 @@ +import assert from "node:assert/strict" +import { readFileSync } from "node:fs" +import { dirname, resolve } from "node:path" +import test from "node:test" +import { fileURLToPath } from "node:url" + +import { parseRunReceipt, ReceiptError, verifyRunReceipt } from "./observation.ts" + +const here = dirname(fileURLToPath(import.meta.url)) +const fixture = readFileSync( + resolve(here, "../../../ir/yield.observation.v1/testdata/run-receipt.canonical.jsonl"), +).subarray(0, -1) + +test("parses and verifies the shared Go canonical receipt", () => { + const receipt = parseRunReceipt(fixture) + assert.equal( + receipt.receipt_digest, + "sha256:a0995d74d51e472f61dc2001f82b93fd8086f9c2f2816a22bb7e5c7fa3c01cdd", + ) + assert.equal(receipt.operations[0].kind, "agent_task") + verifyRunReceipt(receipt, fixture) +}) + +test("rejects unknown fields, noncanonical bytes, and digest mismatch", () => { + const document = JSON.parse(fixture.toString("utf8")) + document.prompt = "secret" + assert.throws(() => parseRunReceipt(JSON.stringify(document)), ReceiptError) + assert.throws(() => parseRunReceipt(Buffer.concat([fixture, Buffer.from("\n")])), /canonical/) + assert.throws( + () => + parseRunReceipt( + fixture.toString("utf8").replace(document.receipt_digest, `sha256:${"0".repeat(64)}`), + ), + /digest/, + ) +}) diff --git a/sdk/typescript/src/observation.ts b/sdk/typescript/src/observation.ts new file mode 100644 index 0000000..46afdc2 --- /dev/null +++ b/sdk/typescript/src/observation.ts @@ -0,0 +1,613 @@ +import { createHash } from "node:crypto" + +export type ReceiptOperationKind = "ask_user" | "agent_task" | "run_command" +export type LifecyclePhase = + | "initializing" + | "initialization_failed" + | "awaiting_response" + | "advancing" + | "recoverable_error" + | "diverged" + | "terminal" + +export interface JournalBinding { + run_id: string + head_sequence: number + head_digest: string +} + +export interface RunIdentity { + id: string + input_digest?: string +} + +export interface ProfileDigest { + profile: string + value: string +} + +export interface ReceiptSkillIdentity { + name: string + version?: string + binding_digest?: string + source_digest?: ProfileDigest +} + +export interface RuntimeIdentity { + supervisor_version?: string + required_version?: string + compatible?: boolean +} + +export interface TimingSummary { + started_at: string + last_observed_at: string + ended_at?: string + elapsed_ms?: number + clock_anomaly?: boolean +} + +export interface OperationObservation { + sequence: number + kind: ReceiptOperationKind + operation_key_digest: string + requested_at: string + completed_at?: string + elapsed_ms?: number + result_digest?: string + clock_anomaly?: boolean +} + +export interface OperationSummary { + kind: ReceiptOperationKind + requested: number + completed: number + total_elapsed_ms: number +} + +export interface OutcomeSummary { + phase: LifecyclePhase + terminal_disposition?: "completed" | "blocked" | "refused" + terminal_cause?: + "completed" | "blocked" | "refused" | "requirement_failed" | "completion_unproven" + result_digest?: string + failure_code?: FailureCode +} + +export type FailureCode = + | "manifest_invalid" + | "manifest_read_failed" + | "runtime_version_missing" + | "runtime_incompatible" + | "runner_missing" + | "source_digest_failed" + | "initialization_failed" + | "invalid_program_output" + | "execution_timeout" + | "subprocess_failed" + | "execution_failed" + | "command_execution_failed" + +export interface RequirementOutcome { + outcome: "passed" | "failed" + claim_digest: string + evidence_digest?: string +} + +export interface ResponseRejectionSummary { + reason: + | "wrong-run" + | "stale-response" + | "duplicate-response" + | "wrong-request" + | "schema-invalid" + | "digest-mismatch" + | "completion-unproven" + | "run-closed" + | "no-pending-operation" + count: number +} + +export interface DivergenceOutcome { + sequence: number + expected_digest: string + got_digest: string +} + +export interface ExperimentContext { + experiment_id: string + cohort_id?: string + variant_id: string + role: "baseline" | "candidate" + baseline_variant_id?: string + parent_skill_version?: string +} + +export interface RunReceipt { + schema: "yield.observation.v1" + kind: "run_receipt" + receipt_digest: string + journal: JournalBinding + run: RunIdentity + skill: ReceiptSkillIdentity + runtime?: RuntimeIdentity + timing: TimingSummary + operations: OperationObservation[] + operation_summaries: OperationSummary[] + outcome: OutcomeSummary + requirements: RequirementOutcome[] + response_rejections: ResponseRejectionSummary[] + divergences: DivergenceOutcome[] + experiment?: ExperimentContext +} + +export class ReceiptError extends Error {} + +type RecordValue = Record + +const digestPattern = /^sha256:[0-9a-f]{64}$/ +const semverPattern = /^\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.-]+)?$/ +const sourceProfilePattern = /^yield\.skill-source\.v[0-9]+$/ +const identifierPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/ +const timestampPattern = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,9})?(?:Z|[+-]\d{2}:\d{2})$/ + +const operationKinds = ["ask_user", "agent_task", "run_command"] as const +const phases = [ + "initializing", + "initialization_failed", + "awaiting_response", + "advancing", + "recoverable_error", + "diverged", + "terminal", +] as const +const dispositions = ["completed", "blocked", "refused"] as const +const terminalCauses = [ + "completed", + "blocked", + "refused", + "requirement_failed", + "completion_unproven", +] as const +const failureCodes = [ + "manifest_invalid", + "manifest_read_failed", + "runtime_version_missing", + "runtime_incompatible", + "runner_missing", + "source_digest_failed", + "initialization_failed", + "invalid_program_output", + "execution_timeout", + "subprocess_failed", + "execution_failed", + "command_execution_failed", +] as const +const rejectionReasons = [ + "wrong-run", + "stale-response", + "duplicate-response", + "wrong-request", + "schema-invalid", + "digest-mismatch", + "completion-unproven", + "run-closed", + "no-pending-operation", +] as const + +function fail(path: string, message: string): never { + throw new ReceiptError(`${path}: ${message}`) +} + +function object( + value: unknown, + path: string, + allowed: readonly string[], + required: readonly string[], +): RecordValue { + if (value === null || typeof value !== "object" || Array.isArray(value)) + fail(path, "must be an object") + const record = value as RecordValue + for (const key of Object.keys(record)) + if (!allowed.includes(key)) fail(path, `unknown field ${JSON.stringify(key)}`) + for (const key of required) + if (!Object.hasOwn(record, key)) fail(path, `missing field ${JSON.stringify(key)}`) + return record +} + +function string(value: unknown, path: string, pattern?: RegExp): string { + if ( + typeof value !== "string" || + value.length === 0 || + (pattern && !pattern.test(value)) || + [...value].some((character) => { + const point = character.codePointAt(0)! + return point >= 0xd800 && point <= 0xdfff + }) + ) + fail(path, "must be a valid non-empty string") + return value +} + +function timestamp(value: unknown, path: string): string { + const text = string(value, path, timestampPattern) + const match = text.match( + /^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(?:\.\d{1,9})?(?:Z|[+-](\d{2}):(\d{2}))$/, + )! + const [year, month, day, hour, minute, second, offsetHour, offsetMinute] = match + .slice(1) + .map((part) => (part === undefined ? 0 : Number(part))) + const leap = year % 4 === 0 && (year % 100 !== 0 || year % 400 === 0) + const days = [0, 31, leap ? 29 : 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31] + if ( + month < 1 || + month > 12 || + day < 1 || + day > days[month] || + hour > 23 || + minute > 59 || + second > 59 || + offsetHour > 23 || + offsetMinute > 59 + ) + fail(path, "must be an RFC 3339 timestamp") + return text +} + +function integer(value: unknown, path: string, minimum: number): number { + if (!Number.isSafeInteger(value) || (value as number) < minimum) + fail(path, `must be an integer >= ${minimum}`) + return value as number +} + +function boolean(value: unknown, path: string): boolean { + if (typeof value !== "boolean") fail(path, "must be a boolean") + return value +} + +function oneOf( + value: unknown, + path: string, + values: T, +): T[number] { + if (typeof value !== "string" || !values.includes(value)) + fail(path, `must be one of ${values.join(", ")}`) + return value as T[number] +} + +function optional( + record: RecordValue, + key: string, + validate: (value: unknown, path: string) => unknown, + path: string, +): void { + if (Object.hasOwn(record, key)) validate(record[key], `${path}.${key}`) +} + +function array( + record: RecordValue, + key: string, + validate: (value: unknown, path: string) => void, + path: string, +): void { + const value = record[key] + if (!Array.isArray(value)) fail(`${path}.${key}`, "must be an array") + value.forEach((item, index) => validate(item, `${path}.${key}[${index}]`)) +} + +function validateReceipt(value: unknown): asserts value is RunReceipt { + const receipt = object( + value, + "$", + [ + "schema", + "kind", + "receipt_digest", + "journal", + "run", + "skill", + "runtime", + "timing", + "operations", + "operation_summaries", + "outcome", + "requirements", + "response_rejections", + "divergences", + "experiment", + ], + [ + "schema", + "kind", + "receipt_digest", + "journal", + "run", + "skill", + "timing", + "operations", + "operation_summaries", + "outcome", + "requirements", + "response_rejections", + "divergences", + ], + ) + if (receipt.schema !== "yield.observation.v1" || receipt.kind !== "run_receipt") + fail("$", "invalid schema or kind") + string(receipt.receipt_digest, "$.receipt_digest", digestPattern) + + const journal = object( + receipt.journal, + "$.journal", + ["run_id", "head_sequence", "head_digest"], + ["run_id", "head_sequence", "head_digest"], + ) + const runID = string(journal.run_id, "$.journal.run_id") + integer(journal.head_sequence, "$.journal.head_sequence", 1) + string(journal.head_digest, "$.journal.head_digest", digestPattern) + + const run = object(receipt.run, "$.run", ["id", "input_digest"], ["id"]) + if (string(run.id, "$.run.id") !== runID) fail("$.run.id", "must match journal.run_id") + optional(run, "input_digest", (v, p) => string(v, p, digestPattern), "$.run") + + const skill = object( + receipt.skill, + "$.skill", + ["name", "version", "binding_digest", "source_digest"], + ["name"], + ) + string(skill.name, "$.skill.name") + optional(skill, "version", (v, p) => string(v, p, semverPattern), "$.skill") + optional(skill, "binding_digest", (v, p) => string(v, p, digestPattern), "$.skill") + if (Object.hasOwn(skill, "source_digest")) { + const source = object( + skill.source_digest, + "$.skill.source_digest", + ["profile", "value"], + ["profile", "value"], + ) + string(source.profile, "$.skill.source_digest.profile", sourceProfilePattern) + string(source.value, "$.skill.source_digest.value", digestPattern) + } + + if (Object.hasOwn(receipt, "runtime")) { + const runtime = object( + receipt.runtime, + "$.runtime", + ["supervisor_version", "required_version", "compatible"], + [], + ) + if (Object.keys(runtime).length === 0) fail("$.runtime", "must not be empty") + optional(runtime, "supervisor_version", (v, p) => string(v, p, semverPattern), "$.runtime") + optional(runtime, "required_version", (v, p) => string(v, p, semverPattern), "$.runtime") + optional(runtime, "compatible", boolean, "$.runtime") + if ( + Object.hasOwn(runtime, "compatible") && + (!Object.hasOwn(runtime, "supervisor_version") || !Object.hasOwn(runtime, "required_version")) + ) + fail("$.runtime.compatible", "requires both runtime versions") + } + + const timing = object( + receipt.timing, + "$.timing", + ["started_at", "last_observed_at", "ended_at", "elapsed_ms", "clock_anomaly"], + ["started_at", "last_observed_at"], + ) + timestamp(timing.started_at, "$.timing.started_at") + timestamp(timing.last_observed_at, "$.timing.last_observed_at") + optional(timing, "ended_at", timestamp, "$.timing") + optional(timing, "elapsed_ms", (v, p) => integer(v, p, 0), "$.timing") + optional(timing, "clock_anomaly", boolean, "$.timing") + if (timing.clock_anomaly === false) + fail("$.timing.clock_anomaly", "false must be omitted from canonical receipts") + + array( + receipt, + "operations", + (item, path) => { + const operation = object( + item, + path, + [ + "sequence", + "kind", + "operation_key_digest", + "requested_at", + "completed_at", + "elapsed_ms", + "result_digest", + "clock_anomaly", + ], + ["sequence", "kind", "operation_key_digest", "requested_at"], + ) + integer(operation.sequence, `${path}.sequence`, 1) + oneOf(operation.kind, `${path}.kind`, operationKinds) + string(operation.operation_key_digest, `${path}.operation_key_digest`, digestPattern) + timestamp(operation.requested_at, `${path}.requested_at`) + optional(operation, "completed_at", timestamp, path) + optional(operation, "elapsed_ms", (v, p) => integer(v, p, 0), path) + optional(operation, "result_digest", (v, p) => string(v, p, digestPattern), path) + optional(operation, "clock_anomaly", boolean, path) + if (operation.clock_anomaly === false) + fail(`${path}.clock_anomaly`, "false must be omitted from canonical receipts") + }, + "$", + ) + + array( + receipt, + "operation_summaries", + (item, path) => { + const summary = object( + item, + path, + ["kind", "requested", "completed", "total_elapsed_ms"], + ["kind", "requested", "completed", "total_elapsed_ms"], + ) + oneOf(summary.kind, `${path}.kind`, operationKinds) + const requested = integer(summary.requested, `${path}.requested`, 0) + const completed = integer(summary.completed, `${path}.completed`, 0) + if (completed > requested) fail(`${path}.completed`, "must not exceed requested") + integer(summary.total_elapsed_ms, `${path}.total_elapsed_ms`, 0) + }, + "$", + ) + + const outcome = object( + receipt.outcome, + "$.outcome", + ["phase", "terminal_disposition", "terminal_cause", "result_digest", "failure_code"], + ["phase"], + ) + const phase = oneOf(outcome.phase, "$.outcome.phase", phases) + optional(outcome, "terminal_disposition", (v, p) => oneOf(v, p, dispositions), "$.outcome") + optional(outcome, "terminal_cause", (v, p) => oneOf(v, p, terminalCauses), "$.outcome") + optional(outcome, "result_digest", (v, p) => string(v, p, digestPattern), "$.outcome") + optional(outcome, "failure_code", (v, p) => oneOf(v, p, failureCodes), "$.outcome") + if (phase === "terminal") { + if ( + !Object.hasOwn(outcome, "terminal_disposition") || + !Object.hasOwn(outcome, "terminal_cause") || + !Object.hasOwn(timing, "ended_at") + ) + fail("$.outcome", "terminal outcome requires disposition, cause, and end time") + } else if ( + Object.hasOwn(outcome, "terminal_disposition") || + Object.hasOwn(outcome, "terminal_cause") + ) + fail("$.outcome", "nonterminal outcome cannot contain terminal fields") + if ( + (phase === "initialization_failed" || phase === "recoverable_error") && + !Object.hasOwn(outcome, "failure_code") + ) + fail("$.outcome.failure_code", "required for failure phase") + + array( + receipt, + "requirements", + (item, path) => { + const requirement = object( + item, + path, + ["outcome", "claim_digest", "evidence_digest"], + ["outcome", "claim_digest"], + ) + oneOf(requirement.outcome, `${path}.outcome`, ["passed", "failed"] as const) + string(requirement.claim_digest, `${path}.claim_digest`, digestPattern) + optional(requirement, "evidence_digest", (v, p) => string(v, p, digestPattern), path) + }, + "$", + ) + + array( + receipt, + "response_rejections", + (item, path) => { + const rejection = object(item, path, ["reason", "count"], ["reason", "count"]) + oneOf(rejection.reason, `${path}.reason`, rejectionReasons) + integer(rejection.count, `${path}.count`, 1) + }, + "$", + ) + + array( + receipt, + "divergences", + (item, path) => { + const divergence = object( + item, + path, + ["sequence", "expected_digest", "got_digest"], + ["sequence", "expected_digest", "got_digest"], + ) + integer(divergence.sequence, `${path}.sequence`, 1) + string(divergence.expected_digest, `${path}.expected_digest`, digestPattern) + string(divergence.got_digest, `${path}.got_digest`, digestPattern) + }, + "$", + ) + + if (Object.hasOwn(receipt, "experiment")) { + const experiment = object( + receipt.experiment, + "$.experiment", + [ + "experiment_id", + "cohort_id", + "variant_id", + "role", + "baseline_variant_id", + "parent_skill_version", + ], + ["experiment_id", "variant_id", "role"], + ) + string(experiment.experiment_id, "$.experiment.experiment_id", identifierPattern) + optional(experiment, "cohort_id", (v, p) => string(v, p, identifierPattern), "$.experiment") + string(experiment.variant_id, "$.experiment.variant_id", identifierPattern) + oneOf(experiment.role, "$.experiment.role", ["baseline", "candidate"] as const) + optional( + experiment, + "baseline_variant_id", + (v, p) => string(v, p, identifierPattern), + "$.experiment", + ) + optional( + experiment, + "parent_skill_version", + (v, p) => string(v, p, semverPattern), + "$.experiment", + ) + } +} + +function canonical(value: unknown): string { + if (value === null) return "null" + if (typeof value === "boolean") return value ? "true" : "false" + if (typeof value === "string") return JSON.stringify(value) + if (typeof value === "number") { + if (!Number.isSafeInteger(value)) + throw new ReceiptError("receipt contains a non-integer or unsafe JSON number") + return String(value) + } + if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]` + if (typeof value === "object") { + const record = value as RecordValue + return `{${Object.keys(record) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonical(record[key])}`) + .join(",")}}` + } + throw new ReceiptError("receipt contains a non-JSON value") +} + +function rawBytes(bytes: Uint8Array | string): Uint8Array { + return typeof bytes === "string" ? Buffer.from(bytes, "utf8") : bytes +} + +/** Verify a typed receipt against its exact canonical bytes and digest. */ +export function verifyRunReceipt(receipt: RunReceipt, bytes: Uint8Array | string): void { + validateReceipt(receipt) + const canonicalReceipt = Buffer.from(canonical(receipt), "utf8") + if (!Buffer.from(rawBytes(bytes)).equals(canonicalReceipt)) + throw new ReceiptError("receipt bytes are not canonical JSON") + const body = { ...receipt } as RecordValue + delete body.receipt_digest + const digest = `sha256:${createHash("sha256").update(canonical(body)).digest("hex")}` + if (digest !== receipt.receipt_digest) + throw new ReceiptError("receipt digest verification failed") +} + +/** Parse and verify one canonical yield.observation.v1 receipt. */ +export function parseRunReceipt(bytes: Uint8Array | string): RunReceipt { + const raw = rawBytes(bytes) + let decoded: unknown + try { + decoded = JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(raw)) + } catch (error) { + throw new ReceiptError(`receipt does not decode: ${String(error)}`) + } + validateReceipt(decoded) + verifyRunReceipt(decoded, raw) + return decoded +}