附录：随书实验完整代码

// plancheck admits one untrusted model proposal. It neither approves nor executes it.
package main

import (
    "encoding/json"
    "fmt"
    "github.com/Aias00/durable-agent-workflow-book/lab"
    "io"
    "os"
)

func main() {
    raw, err := io.ReadAll(io.LimitReader(os.Stdin, 4097))
    if err != nil {
        fmt.Fprintln(os.Stderr, "cannot read bounded proposal")
        os.Exit(1)
    }
    p, err := lab.AdmitProposal(raw, lab.ToolContract{Name: "credit-service", Idempotency: true, AuthoritativeLookup: true})
    if err != nil {
        fmt.Fprintln(os.Stderr, "proposal rejected")
        os.Exit(1)
    }
    _ = json.NewEncoder(os.Stdout).Encode(map[string]any{"admitted": true, "fingerprint": p.Fingerprint(), "plan": p, "next_required_stage": "persist_then_wait_for_approval"})
}

// worker is an actual subprocess used by the crash-recovery experiments.
package main

import (
    "bufio"
    "context"
    "flag"
    "fmt"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/Aias00/durable-agent-workflow-book/lab"
)

func main() {
    schema := flag.String("schema", "", "isolated schema created by the experiment")
    endpoint := flag.String("tool", "", "loopback tool endpoint")
    name := flag.String("name", "worker-process", "worker owner")
    pause := flag.String("pause", "", "pause after_claim or after_tool until stdin newline (or SIGKILL)")
    gate := flag.Bool("start-gate", false, "wait for stdin newline before claiming")
    reconcile := flag.String("reconcile", "", "query this unknown workflow before running one ready item")
    flag.Parse()
    ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer cancel()
    input := bufio.NewReader(os.Stdin)
    waitGate := func() {
        ready := make(chan struct{})
        go func() { _, _ = input.ReadString('\n'); close(ready) }()
        select {
        case <-ready:
        case <-ctx.Done():
        }
    }
    store, err := lab.Open(ctx, *schema, false)
    if err != nil {
        fmt.Fprintln(os.Stderr, err)
        os.Exit(2)
    }
    defer store.Close()
    tool, err := lab.NewToolClient(*endpoint)
    if err != nil {
        fmt.Fprintln(os.Stderr, err)
        os.Exit(2)
    }
    defer tool.Close()
    worker := lab.Worker{Store: store, Tool: tool, Name: *name, Lease: 3 * time.Second, HeartbeatEvery: 500 * time.Millisecond, Backoff: 100 * time.Millisecond}
    worker.Hook = func(stage string, w lab.Workflow) {
        fmt.Printf("BOOK_STAGE %s %s %d\n", stage, w.ID, w.Fence)
        if *pause == stage {
            waitGate()
        }
    }
    if *gate {
        fmt.Println("BOOK_READY")
        waitGate()
    }
    if *reconcile != "" {
        err = worker.Reconcile(ctx, *reconcile)
    }
    if err == nil {
        err = worker.RunOne(ctx)
    }
    if err != nil {
        fmt.Fprintln(os.Stderr, err)
        os.Exit(1)
    }
    fmt.Println("BOOK_DONE")
}

#!/bin/sh
# Download only fixed official assets; binaries and weights remain in ignored .tools.
set -eu
task_repo_root=$(CDPATH= cd -- "$(dirname -- "$0")/../.." && pwd)
cd "$task_repo_root"
mkdir -p .tools/model
case "$(uname -s):$(uname -m)" in
 Darwin:arm64) task_archive=llama-b11789-bin-macos-arm64.tar.gz; task_archive_sha=a85c325b80b3ce0eab2481db4e0ec988386fc0c044c95a0e6c2efea8c6842606 ;;
 Linux:x86_64) task_archive=llama-b11789-bin-ubuntu-x64.tar.gz; task_archive_sha=d5ec57395e369cec744849a5fd629455c814b0c4a0889eba55fca7de054c0636 ;;
 *) echo 'This fixture supports macOS arm64 and Linux x86_64 only.' >&2; exit 1 ;;
esac
verify_file() {
 python3 - "$1" "$2" <<'PY'
import hashlib,sys
from pathlib import Path
p=Path(sys.argv[1])
if not p.exists():sys.exit(1)
with p.open('rb') as f:digest=hashlib.file_digest(f,'sha256').hexdigest()
sys.exit(0 if digest==sys.argv[2] else 1)
PY
}
task_tar=.tools/model/llama-pinned.tar.gz
if ! verify_file "$task_tar" "$task_archive_sha"; then
 curl --fail --location --silent --show-error --retry 3 --max-time 180 "https://github.com/ggml-org/llama.cpp/releases/download/b11789/$task_archive" -o "$task_tar.tmp"
 verify_file "$task_tar.tmp" "$task_archive_sha"
 mv "$task_tar.tmp" "$task_tar"
fi
task_gguf=.tools/model/qwen2.5-0.5b-instruct-q4_k_m.gguf
task_gguf_sha=74a4da8c9fdbcd15bd1f6d01d621410d31c6fc00986f5eb687824e7b93d7a9db
if ! verify_file "$task_gguf" "$task_gguf_sha"; then
 curl --fail --location --silent --show-error --retry 3 --max-time 300 'https://huggingface.co/Qwen/Qwen2.5-0.5B-Instruct-GGUF/resolve/9217f5db79a29953eb74d5343926648285ec7e67/qwen2.5-0.5b-instruct-q4_k_m.gguf' -o "$task_gguf.tmp"
 verify_file "$task_gguf.tmp" "$task_gguf_sha"
 mv "$task_gguf.tmp" "$task_gguf"
fi
python3 - "$task_tar" <<'PY'
from pathlib import Path
import tarfile,sys
with tarfile.open(sys.argv[1]) as tar:tar.extractall('.tools/model/runtime',filter='data')
files=list(Path('.tools/model/runtime').rglob('llama-cli'))
assert len(files)==1
print('Verified binary:',files[0])
print('Verified GGUF: .tools/model/qwen2.5-0.5b-instruct-q4_k_m.gguf')
PY

{"type":"object","properties":{"version":{"const":1},"action":{"const":"issue_credit"},"account":{"type":"string","enum":["account-a","account-b"]},"amount_minor":{"type":"integer","minimum":1,"maximum":10000},"currency":{"const":"CNY"}},"required":["version","action","account","amount_minor","currency"],"additionalProperties":false}

#!/usr/bin/env python3
"""Run a pinned real GGUF model, then admit its untrusted output through Go.
This tiny authored fixture is not a business-quality benchmark.
"""
from pathlib import Path
import argparse,hashlib,json,platform,subprocess,time
from datetime import datetime,timezone
ROOT=Path(__file__).resolve().parents[2]
MODEL_SHA='74a4da8c9fdbcd15bd1f6d01d621410d31c6fc00986f5eb687824e7b93d7a9db'
REVISION='9217f5db79a29953eb74d5343926648285ec7e67'
parser=argparse.ArgumentParser();parser.add_argument('--binary',required=True);parser.add_argument('--model',required=True);parser.add_argument('--output',default='book/model-experiment.json');args=parser.parse_args()
model=Path(args.model).resolve();binary=Path(args.binary).resolve()
assert hashlib.file_digest(model.open('rb'),'sha256').hexdigest()==MODEL_SHA
version=subprocess.run([str(binary),'--version'],capture_output=True,text=True,check=True,timeout=30)
assert '11789' in version.stdout+version.stderr
cases=[('account-a',500,'Create a plan to issue 500 CNY minor units to account-a.'),('account-b',1200,'Create a plan to issue 1200 CNY minor units to account-b.')]
results=[]
for account,amount,prompt in cases:
 argv=[str(binary),'-m',str(model),'-ngl','0','--device','none','-t','2','-c','2048','-n','180','--temp','0','--seed','42','--single-turn','--simple-io','--no-display-prompt','--no-show-timings','--color','off','--log-disable','--no-warmup','-sys','Output exactly one JSON object, no explanations. Use these exact keys: version, action, account, amount_minor, currency. version=1, action=issue_credit, currency=CNY. Copy account and amount_minor from the user request. Example shape: {"version":1,"action":"issue_credit","account":"account-a","amount_minor":1,"currency":"CNY"}. Never add approvals, identity or tool capabilities.','-p',prompt]
 started=time.monotonic();run=subprocess.run(argv,capture_output=True,text=True,timeout=180,check=True)
 output=run.stdout;objects=[]
 for pos,char in enumerate(output):
  if char=='{':
   try:
    obj,end=json.JSONDecoder().raw_decode(output[pos:])
    if isinstance(obj,dict) and 'action' in obj:objects.append(obj)
   except json.JSONDecodeError:pass
 assert len(objects)==1,('expected one actual generated proposal',output[-1500:])
 proposal=objects[0]
 trace=ROOT/'.build/model-traces';trace.mkdir(parents=True,exist_ok=True);(trace/(account+'.stdout')).write_text(output);(trace/(account+'.stderr')).write_text(run.stderr)
 admitted=subprocess.run(['go','run','./cmd/plancheck'],input=json.dumps(proposal),text=True,capture_output=True,check=False,cwd=ROOT,timeout=60)
 check=json.loads(admitted.stdout) if admitted.returncode==0 else {'admitted':False,'state':'rejected','reason':'strict Go admission rejected generated proposal'}
 matched=proposal.get('account')==account and proposal.get('amount_minor')==amount
 results.append({'prompt':prompt,'proposal':proposal,'expected_account':account,'expected_amount_minor':amount,'matches_authored_request':matched,'admission':check,'elapsed_seconds':round(time.monotonic()-started,3),'generated_stdout':output,'generated_stdout_sha256':hashlib.sha256(output.encode()).hexdigest()})
 print(json.dumps(results[-1],ensure_ascii=False),flush=True)
report={'tested_at_utc':datetime.now(timezone.utc).isoformat(),'model':'Qwen/Qwen2.5-0.5B-Instruct-GGUF','revision':REVISION,'model_file':model.name,'model_sha256':MODEL_SHA,'model_bytes':model.stat().st_size,'license':'Apache-2.0','engine_release':'b11789','engine_version':(version.stdout+version.stderr).strip(),'engine_binary_sha256':hashlib.file_digest(binary.open('rb'),'sha256').hexdigest(),'platform':platform.platform(),'cpu_only':True,'threads':2,'temperature':0,'seed':42,'schema_constrained':False,'setup_note':'The initial constrained-schema attempt failed in llama.cpp b11789 sampler initialization; this report uses unconstrained model generation and independent strict Go admission.','cases':results,'limitations':['Two author-constructed requests, not a business-quality benchmark or model ranking.','Schema constrains syntax and permitted fields, not semantic correctness.','The CLI wrapper stdout is scanned for exactly one candidate object; only that candidate is sent to strict Go admission.','Go admission is stateless: it does not persist a workflow, approve or execute a business effect.','Workflow crash experiments are separate deterministic tests against real PostgreSQL and HTTP.']}
out=ROOT/args.output;out.parent.mkdir(parents=True,exist_ok=True);out.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
# Semantic mismatches and admission rejects remain visible; never discard them to improve a score.
print('RECORDED: actual CPU GGUF generation and strict Go admission; no business effect executed')

#!/usr/bin/env python3
"""Turn raw go test -json into evidence; fail publication if integration skipped."""
import hashlib
import re
import json
import os
import pathlib
import subprocess
import sys
from datetime import datetime, timezone

folder = pathlib.Path(sys.argv[1] if len(sys.argv) > 1 else "evidence")
tests = {}
observations = []
package_failures = []
for line in (folder / "tests.jsonl").read_text().splitlines():
    item = json.loads(line)
    name = item.get("Test")
    if name and item["Action"] in ("pass", "fail", "skip"):
        tests[name] = item["Action"]
    if item["Action"] == "fail" and not name:
        package_failures.append(item["Package"])
    output = item.get("Output", "")
    marker = "BOOK_OBSERVATION "
    if marker in output:
        observations.append({"test": name, "observation": json.loads(output.split(marker, 1)[1])})

expected = sorted(re.findall(r"^func (Test\w+)\(t \*testing.T\)", pathlib.Path("lab/lab_test.go").read_text(), re.MULTILINE))
missing = sorted(set(expected) - set(tests))
passed = len(expected) >= 26 and not missing and all(v == "pass" for v in tests.values()) and not package_failures and bool(os.environ.get("BOOK_LAB_DSN"))
result = {
    "schema_version": 1,
    "status": "PASS" if passed else "FAIL",
    "timestamp_utc": datetime.now(timezone.utc).isoformat(),
    "commit": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(),
    "run_url": os.environ.get("BOOK_RUN_URL"),
    "go": subprocess.check_output(["go", "version"], text=True).strip(),
    "postgresql": (folder / "postgresql-version.txt").read_text().strip(),
    "postgres_image_digest": (folder / "postgres-image-digest.txt").read_text().strip(),
    "command": "go test -race -count=1 -json ./...",
    "integration_dsn_configured": bool(os.environ.get("BOOK_LAB_DSN")),
    "tests": tests,
    "expected_top_level_tests": expected,
    "missing_tests": missing,
    "source_sha256": {
        str(p): hashlib.sha256(p.read_bytes()).hexdigest()
        for p in sorted([*pathlib.Path("lab").glob("*.go"), *pathlib.Path("lab").glob("*.sql"), *pathlib.Path("cmd").rglob("*.go"), pathlib.Path("go.mod"), pathlib.Path("go.sum")])
    },
    "package_failures": package_failures,
    "observations": observations,
    "limits": [
        "One side-effect node; not a general workflow engine or exactly-once system.",
        "Workflow and tool ledger share one PostgreSQL instance but use separate transactions.",
        "SIGKILL experiments wait for the actual three-second database lease; other boundary tests label SQL-injected expiry.",
        "Loopback development credentials, authoritative lab lookup, no multi-cluster availability or provider-authentication claim.",
    ],
}
(folder / "runtime.json").write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n")
print(json.dumps({"status": result["status"], "tests": len(tests), "observations": len(observations), "skips": sum(v == "skip" for v in tests.values())}))
sys.exit(0 if passed else 1)

package lab

import (
    "bufio"
    "bytes"
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "net/http"
    "net/http/httptest"
    "os"
    "os/exec"
    "path/filepath"
    "strings"
    "sync"
    "testing"
    "time"
)

var workerBinary string

func TestMain(m *testing.M) {
    cleanup := func() {}
    if os.Getenv("BOOK_LAB_DSN") != "" {
        dir, err := os.MkdirTemp("", "workflow-worker-")
        if err != nil {
            panic(err)
        }
        cleanup = func() { os.RemoveAll(dir) }
        workerBinary = filepath.Join(dir, "worker")
        command := exec.Command("go", "build", "-race", "-o", workerBinary, "../cmd/worker")
        if output, err := command.CombinedOutput(); err != nil {
            fmt.Fprintln(os.Stderr, string(output))
            cleanup()
            os.Exit(2)
        }
    }
    code := m.Run()
    cleanup()
    os.Exit(code)
}
func must(t *testing.T, err error) {
    t.Helper()
    if err != nil {
        t.Fatal(err)
    }
}
func db(t *testing.T) *Store {
    t.Helper()
    if os.Getenv("BOOK_LAB_DSN") == "" {
        t.Skip("real PostgreSQL integration: set BOOK_LAB_DSN or use repository CI")
    }
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    s, err := Open(ctx, "wf_"+ID(), true)
    must(t, err)
    t.Cleanup(func() { must(t, s.Close()) })
    return s
}

var proposer = Actor{Tenant: "tenant-a", Subject: "employee-a", Role: "proposer"}
var approver = Actor{Tenant: "tenant-a", Subject: "manager-a", Role: "approver"}

func sample(idempotent bool) Plan {
    return Plan{Version: 1, Action: "issue_credit", Account: "account-a", AmountMinor: 500, Currency: "CNY", Idempotent: idempotent}
}
func approved(t *testing.T, s *Store, idempotent bool) Workflow {
    t.Helper()
    ctx := context.Background()
    w, err := s.Create(ctx, proposer, ID(), sample(idempotent))
    must(t, err)
    must(t, s.Approve(ctx, approver, w.ID, w.Fingerprint, time.Minute))
    w, err = s.Get(ctx, w.ID)
    must(t, err)
    return w
}
func effects(t *testing.T, s *Store) int {
    t.Helper()
    var n int
    must(t, s.DB.QueryRow("SELECT count(*) FROM tool_effects").Scan(&n))
    return n
}
func state(t *testing.T, s *Store, id string) Workflow {
    t.Helper()
    w, err := s.Get(context.Background(), id)
    must(t, err)
    return w
}
func observe(t *testing.T, value any) {
    t.Helper()
    b, err := json.Marshal(value)
    must(t, err)
    t.Log("BOOK_OBSERVATION " + string(b))
}
func fixture(t *testing.T, s *Store, idempotent bool) (*ToolService, *ToolClient, *Worker) {
    t.Helper()
    server := NewTool(s.DB, idempotent)
    httpServer := httptest.NewServer(server)
    t.Cleanup(httpServer.Close)
    client, err := NewToolClient(httpServer.URL)
    must(t, err)
    t.Cleanup(client.Close)
    return server, client, &Worker{Store: s, Tool: client, Name: "worker-a", Lease: 3 * time.Second, Backoff: 20 * time.Millisecond}
}
func expired(t *testing.T, s *Store, id string) {
    t.Helper()
    _, err := s.DB.Exec("UPDATE workflows SET lease_until=clock_timestamp()-interval '1 second' WHERE id=$1", id)
    must(t, err)
}
func due(t *testing.T, s *Store, id string) {
    t.Helper()
    _, err := s.DB.Exec("UPDATE workflows SET ready_at=clock_timestamp()-interval '1 second' WHERE id=$1", id)
    must(t, err)
}
func until(t *testing.T, budget time.Duration, f func() bool) {
    t.Helper()
    deadline := time.Now().Add(budget)
    for time.Now().Before(deadline) {
        if f() {
            return
        }
        time.Sleep(20 * time.Millisecond)
    }
    t.Fatal("bounded condition did not become true")
}
func TestProposalCannotGrantAuthority(t *testing.T) {
    base := `{"version":1,"action":"issue_credit","account":"account-a","amount_minor":500,"currency":"CNY"}`
    contract := ToolContract{Name: "credit-service", Idempotency: true, AuthoritativeLookup: true}
    p, err := AdmitProposal([]byte(base), contract)
    must(t, err)
    if !p.Idempotent {
        t.Fatal("capability must come from trusted registry")
    }
    for _, extra := range []string{`"approved":true`, `"idempotent":true`, `"tenant":"victim"`, `"endpoint":"http://evil.invalid"`, `"actor":"admin"`} {
        _, err = AdmitProposal([]byte(strings.TrimSuffix(base, "}")+","+extra+"}"), contract)
        if err == nil {
            t.Fatal("accepted model authority", extra)
        }
    }
    if _, err = AdmitProposal([]byte(base+base), contract); err == nil {
        t.Fatal("accepted two proposals")
    }
    if _, err = AdmitProposal([]byte(strings.Repeat(" ", 4097)), contract); err == nil {
        t.Fatal("unbounded proposal")
    }
    observe(t, map[string]any{"authority_fields_rejected": 5, "capability_from_registry": p.Idempotent})
}
func TestOperationKeyHasUnambiguousTenantScope(t *testing.T) {
    a := Workflow{Tenant: "a:b", Operation: "c"}
    b := Workflow{Tenant: "a", Operation: "b:c"}
    if a.EffectKey() == b.EffectKey() {
        t.Fatal("ambiguous compound identity")
    }
    if a.EffectKey() != (Workflow{Tenant: "a:b", Operation: "c"}).EffectKey() {
        t.Fatal("unstable identity")
    }
    observe(t, map[string]any{"tenant_scope_unambiguous": true, "operation_key_stable": true})
}
func TestOperationIdentityRejectsParameterDrift(t *testing.T) {
    s := db(t)
    ctx := context.Background()
    a, err := s.Create(ctx, proposer, "same-operation", sample(true))
    must(t, err)
    b, err := s.Create(ctx, proposer, "same-operation", sample(true))
    must(t, err)
    if a.ID != b.ID {
        t.Fatal("duplicate workflow")
    }
    changed := sample(true)
    changed.AmountMinor = 501
    if _, err = s.Create(ctx, proposer, "same-operation", changed); !errors.Is(err, ErrConflict) {
        t.Fatal("parameter drift accepted", err)
    }
    observe(t, map[string]any{"same_workflow": a.ID == b.ID, "drift_rejected": true, "state": state(t, s, a.ID).State})
}
func TestApprovalBindsTenantRoleAndFingerprint(t *testing.T) {
    s := db(t)
    ctx := context.Background()
    w, err := s.Create(ctx, proposer, ID(), sample(true))
    must(t, err)
    for _, a := range []Actor{{Tenant: "tenant-b", Subject: "x", Role: "approver"}, proposer} {
        if !errors.Is(s.Approve(ctx, a, w.ID, w.Fingerprint, time.Minute), ErrDenied) {
            t.Fatal("unauthorized approval")
        }
    }
    if !errors.Is(s.Approve(ctx, approver, w.ID, "different-fingerprint", time.Minute), ErrDenied) {
        t.Fatal("unbound approval")
    }
    must(t, s.Approve(ctx, approver, w.ID, w.Fingerprint, time.Minute))
    if state(t, s, w.ID).State != "ready" {
        t.Fatal("approval not persisted")
    }
    observe(t, map[string]any{"wrong_tenant_rejected": true, "wrong_role_rejected": true, "wrong_fingerprint_rejected": true, "state": "ready"})
}
func TestTwoClaimersCannotOwnOneWorkflow(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    start := make(chan struct{})
    var wg sync.WaitGroup
    var lock sync.Mutex
    claims := []Workflow{}
    failures := []error{}
    for _, name := range []string{"a", "b"} {
        wg.Add(1)
        go func(owner string) {
            defer wg.Done()
            <-start
            w, err := s.Claim(context.Background(), owner, time.Second)
            lock.Lock()
            defer lock.Unlock()
            if err != nil {
                failures = append(failures, err)
            }
            if w != nil {
                claims = append(claims, *w)
            }
        }(name)
    }
    close(start)
    wg.Wait()
    if len(failures) > 0 || len(claims) != 1 || claims[0].ID != job.ID {
        t.Fatal(claims, failures)
    }
    observe(t, map[string]any{"concurrent_claimers": 2, "successful_claims": len(claims), "fence": claims[0].Fence})
}
func TestHeartbeatCannotReviveAnExpiredLease(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    w, err := s.Claim(context.Background(), "owner", time.Second)
    must(t, err)
    before := w.LeaseUntil
    must(t, s.Heartbeat(context.Background(), *w, 2*time.Second))
    after := state(t, s, job.ID).LeaseUntil
    if !after.After(*before) {
        t.Fatal("heartbeat did not extend lease")
    }
    expired(t, s, job.ID)
    if !errors.Is(s.Heartbeat(context.Background(), *w, time.Second), ErrStale) {
        t.Fatal("expired owner revived")
    }
    observe(t, map[string]any{"live_lease_extended": true, "expired_owner_rejected": true, "lease_expiry": "SQL-injected"})
}
func TestExpiredOwnerCannotCommitAnOldResult(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    _, client, _ := fixture(t, s, true)
    first, err := s.Claim(context.Background(), "old", time.Second)
    must(t, err)
    expired(t, s, job.ID)
    n, err := s.RecoverExpired(context.Background())
    must(t, err)
    if n != 1 {
        t.Fatal(n)
    }
    must(t, (&Worker{Store: s, Tool: client}).Reconcile(context.Background(), job.ID))
    second, err := s.Claim(context.Background(), "new", time.Second)
    must(t, err)
    receipt, err := client.Call(context.Background(), *second)
    must(t, err)
    if err = s.Complete(context.Background(), *first, receipt); !errors.Is(err, ErrStale) {
        t.Fatal("stale owner accepted", err)
    }
    again, err := client.Call(context.Background(), *second)
    must(t, err)
    must(t, s.Complete(context.Background(), *second, again))
    if again.ID != receipt.ID || effects(t, s) != 1 {
        t.Fatal("duplicate effect")
    }
    observe(t, map[string]any{"old_fence": first.Fence, "new_fence": second.Fence, "effects": effects(t, s), "state": state(t, s, job.ID).State})
}
func TestCommitReplyLostBecomesUnknownThenReconciled(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, _, worker := fixture(t, s, true)
    server.DropResponses.Store(true)
    must(t, worker.RunOne(context.Background()))
    unknown := state(t, s, job.ID)
    if unknown.State != "unknown" || effects(t, s) != 1 {
        t.Fatal(unknown.State)
    }
    must(t, worker.Reconcile(context.Background(), job.ID))
    final := state(t, s, job.ID)
    if final.State != "succeeded" || effects(t, s) != 1 {
        t.Fatal(final.State)
    }
    observe(t, map[string]any{"after_reply_loss": unknown.State, "after_lookup": final.State, "effects": effects(t, s), "post_calls": server.Calls.Load()})
}
func TestNonIdempotentUnknownNeverAutomaticallyReplays(t *testing.T) {
    s := db(t)
    job := approved(t, s, false)
    server, _, worker := fixture(t, s, false)
    server.DropResponses.Store(true)
    must(t, worker.RunOne(context.Background()))
    if state(t, s, job.ID).State != "unknown" || effects(t, s) != 1 {
        t.Fatal("missing unknown effect")
    }
    if !errors.Is(worker.Reconcile(context.Background(), job.ID), ErrManual) {
        t.Fatal("unqueryable tool automatically resolved")
    }
    for i := 0; i < 3; i++ {
        must(t, worker.RunOne(context.Background()))
    }
    if effects(t, s) != 1 || server.Calls.Load() != 1 {
        t.Fatal("unknown replayed")
    }
    observe(t, map[string]any{"state": state(t, s, job.ID).State, "effects": effects(t, s), "post_calls": server.Calls.Load()})
}

// process owns an OS process, not a goroutine pretending to crash a worker.
type process struct {
    cmd    *exec.Cmd
    input  io.WriteCloser
    lines  chan string
    done   chan struct{}
    err    error
    stderr bytes.Buffer
}

func child(t *testing.T, s *Store, c *ToolClient, name string, extra ...string) *process {
    t.Helper()
    args := append([]string{"-schema", s.Schema, "-tool", c.Base, "-name", name}, extra...)
    p := &process{cmd: exec.Command(workerBinary, args...), lines: make(chan string, 16), done: make(chan struct{})}
    p.cmd.Env = append(os.Environ(), "GORACE=atexit_sleep_ms=0")
    var err error
    p.input, err = p.cmd.StdinPipe()
    must(t, err)
    output, err := p.cmd.StdoutPipe()
    must(t, err)
    p.cmd.Stderr = &p.stderr
    must(t, p.cmd.Start())
    scanned := make(chan struct{})
    go func() {
        defer close(scanned)
        scanner := bufio.NewScanner(output)
        for scanner.Scan() {
            p.lines <- scanner.Text()
        }
        close(p.lines)
    }()
    go func() { <-scanned; p.err = p.cmd.Wait(); close(p.done) }()
    t.Cleanup(func() {
        select {
        case <-p.done:
        default:
            _ = p.cmd.Process.Kill()
            <-p.done
        }
        _ = p.input.Close()
    })
    return p
}
func (p *process) stage(t *testing.T, prefix string) string {
    t.Helper()
    timer := time.NewTimer(15 * time.Second)
    defer timer.Stop()
    for {
        select {
        case line, ok := <-p.lines:
            if !ok {
                <-p.done
                t.Fatalf("process ended before %q: %v %s", prefix, p.err, p.stderr.String())
            }
            if strings.HasPrefix(line, prefix) {
                return line
            }
        case <-timer.C:
            t.Fatalf("process timeout waiting for %s", prefix)
        }
    }
}
func (p *process) release(t *testing.T) {
    t.Helper()
    _, err := io.WriteString(p.input, "\n")
    must(t, err)
}
func (p *process) finish(t *testing.T) {
    t.Helper()
    select {
    case <-p.done:
        if p.err != nil {
            t.Fatalf("worker failed: %v %s", p.err, p.stderr.String())
        }
    case <-time.After(15 * time.Second):
        t.Fatal("worker did not exit")
    }
}
func (p *process) kill(t *testing.T) {
    t.Helper()
    must(t, p.cmd.Process.Kill())
    <-p.done
    if p.err == nil {
        t.Fatal("SIGKILL did not terminate worker")
    }
}
func recoverByElapsedLease(t *testing.T, s *Store, id string) {
    t.Helper()
    // Polling the real PostgreSQL clock lets the actual three-second lease expire.
    until(t, 5*time.Second, func() bool {
        var expired bool
        must(t, s.DB.QueryRow("SELECT lease_until<=clock_timestamp() FROM workflows WHERE id=$1", id).Scan(&expired))
        return expired
    })
    n, err := s.RecoverExpired(context.Background())
    must(t, err)
    if n != 1 {
        t.Fatalf("recovered %d", n)
    }
    if state(t, s, id).State != "unknown" {
        t.Fatal("expired lease skipped unknown")
    }
}
func TestTwoWorkerProcessesClaimExactlyOne(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, client, _ := fixture(t, s, true)
    a := child(t, s, client, "process-a", "-start-gate", "-pause", "after_claim")
    b := child(t, s, client, "process-b", "-start-gate", "-pause", "after_claim")
    a.stage(t, "BOOK_READY")
    b.stage(t, "BOOK_READY")
    a.release(t)
    b.release(t)
    until(t, 5*time.Second, func() bool { return state(t, s, job.ID).State == "running" })
    current := state(t, s, job.ID)
    winner, loser := a, b
    if current.Owner == "process-b" {
        winner, loser = b, a
    }
    winner.stage(t, "BOOK_STAGE after_claim")
    loser.finish(t)
    winner.release(t)
    winner.finish(t)
    var claims int
    must(t, s.DB.QueryRow("SELECT count(*) FROM events WHERE kind='claimed'").Scan(&claims))
    if claims != 1 || effects(t, s) != 1 || server.Calls.Load() != 1 || state(t, s, job.ID).State != "succeeded" {
        t.Fatal("duplicate process execution")
    }
    observe(t, map[string]any{"os_processes": 2, "claims": claims, "effects": effects(t, s), "state": state(t, s, job.ID).State})
}
func TestSIGKILLAfterClaimQueries404AndRestarts(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, client, _ := fixture(t, s, true)
    p := child(t, s, client, "crashed", "-pause", "after_claim")
    p.stage(t, "BOOK_STAGE after_claim")
    p.kill(t)
    if effects(t, s) != 0 {
        t.Fatal("effect before dispatch")
    }
    recoverByElapsedLease(t, s, job.ID)
    next := child(t, s, client, "restarted", "-reconcile", job.ID)
    next.finish(t)
    w := state(t, s, job.ID)
    if w.State != "succeeded" || w.Attempt != 2 || effects(t, s) != 1 {
        t.Fatal(w)
    }
    observe(t, map[string]any{"signal": "SIGKILL", "window": "after_claim", "lease_expiry": "elapsed PostgreSQL clock; 3s lease", "state": w.State, "attempts": w.Attempt, "effects": effects(t, s), "post_calls": server.Calls.Load()})
}
func TestSIGKILLAfterToolReconcilesExistingEffect(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, client, _ := fixture(t, s, true)
    p := child(t, s, client, "crashed", "-pause", "after_tool")
    p.stage(t, "BOOK_STAGE after_tool")
    p.kill(t)
    if effects(t, s) != 1 || state(t, s, job.ID).State != "running" {
        t.Fatal("wrong crash window")
    }
    recoverByElapsedLease(t, s, job.ID)
    next := child(t, s, client, "restarted", "-reconcile", job.ID)
    next.finish(t)
    w := state(t, s, job.ID)
    if w.State != "succeeded" || w.Attempt != 1 || effects(t, s) != 1 || server.Calls.Load() != 1 {
        t.Fatal(w)
    }
    observe(t, map[string]any{"signal": "SIGKILL", "window": "after_tool", "lease_expiry": "elapsed PostgreSQL clock; 3s lease", "state": w.State, "attempts": w.Attempt, "effects": effects(t, s), "post_calls": server.Calls.Load()})
}
func TestSIGKILLNonIdempotentRequiresManualReview(t *testing.T) {
    for _, window := range []string{"after_claim", "after_tool"} {
        t.Run(window, func(t *testing.T) {
            s := db(t)
            job := approved(t, s, false)
            server, client, worker := fixture(t, s, false)
            p := child(t, s, client, "crashed", "-pause", window)
            p.stage(t, "BOOK_STAGE "+window)
            p.kill(t)
            recoverByElapsedLease(t, s, job.ID)
            if !errors.Is(worker.Reconcile(context.Background(), job.ID), ErrManual) {
                t.Fatal("unsafe reconciliation")
            }
            next := child(t, s, client, "restarted")
            next.finish(t)
            want := 0
            if window == "after_tool" {
                want = 1
            }
            if state(t, s, job.ID).State != "unknown" || effects(t, s) != want || server.Calls.Load() != int64(want) {
                t.Fatal("non-idempotent replay")
            }
            observe(t, map[string]any{"signal": "SIGKILL", "window": window, "lease_expiry": "elapsed PostgreSQL clock; 3s lease", "state": "unknown", "effects": effects(t, s), "post_calls": server.Calls.Load(), "manual_review": true})
        })
    }
}
func TestApprovalExpiresWhileQueued(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, _, worker := fixture(t, s, true)
    _, err := s.DB.Exec("UPDATE workflows SET approval_expires=clock_timestamp()-interval '1 second' WHERE id=$1", job.ID)
    must(t, err)
    must(t, worker.RunOne(context.Background()))
    if state(t, s, job.ID).State != "waiting_approval" || server.Calls.Load() != 0 {
        t.Fatal("expired approval dispatched")
    }
    observe(t, map[string]any{"expiry": "SQL-injected before claim", "state": state(t, s, job.ID).State, "post_calls": server.Calls.Load()})
}
func TestApprovalExpiresAfterClaim(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, _, worker := fixture(t, s, true)
    worker.Hook = func(stage string, w Workflow) {
        if stage == "after_claim" {
            _, err := s.DB.Exec("UPDATE workflows SET approval_expires=clock_timestamp()-interval '1 second' WHERE id=$1", w.ID)
            must(t, err)
        }
    }
    must(t, worker.RunOne(context.Background()))
    if state(t, s, job.ID).State != "waiting_approval" || server.Calls.Load() != 0 {
        t.Fatal("dispatch ignored approval expiry")
    }
    observe(t, map[string]any{"expiry": "SQL-injected after claim", "state": state(t, s, job.ID).State, "effects": effects(t, s)})
}
func TestCancelBeforeDispatch(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, _, worker := fixture(t, s, true)
    worker.Hook = func(stage string, w Workflow) {
        if stage == "after_claim" {
            must(t, s.Cancel(context.Background(), proposer, w.ID))
        }
    }
    must(t, worker.RunOne(context.Background()))
    if state(t, s, job.ID).State != "cancelled" || server.Calls.Load() != 0 {
        t.Fatal("cancelled work dispatched")
    }
    queued := approved(t, s, true)
    must(t, s.Cancel(context.Background(), proposer, queued.ID))
    must(t, worker.RunOne(context.Background()))
    if state(t, s, queued.ID).State != "cancelled" || server.Calls.Load() != 0 {
        t.Fatal("queued cancellation ignored")
    }
    observe(t, map[string]any{"before_dispatch": "cancelled", "queued": "cancelled", "effects": effects(t, s)})
}
func TestCancelAfterEffectPreservesSucceededFact(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    _, _, worker := fixture(t, s, true)
    worker.Hook = func(stage string, w Workflow) {
        if stage == "after_tool" {
            must(t, s.Cancel(context.Background(), proposer, w.ID))
        }
    }
    must(t, worker.RunOne(context.Background()))
    w := state(t, s, job.ID)
    if w.State != "succeeded" || !w.CancelRequested || effects(t, s) != 1 {
        t.Fatal("cancellation erased an effect")
    }
    observe(t, map[string]any{"state": w.State, "cancel_requested": w.CancelRequested, "effects": effects(t, s), "compensated": false})
}
func TestTrusted429UsesTimedRetry(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, _, worker := fixture(t, s, true)
    worker.Backoff = 250 * time.Millisecond
    server.RejectRemaining.Store(1)
    must(t, worker.RunOne(context.Background()))
    if state(t, s, job.ID).State != "ready" || effects(t, s) != 0 {
        t.Fatal("429 not safely queued")
    }
    must(t, worker.RunOne(context.Background()))
    if server.Calls.Load() != 1 {
        t.Fatal("backoff ignored")
    }
    until(t, 2*time.Second, func() bool {
        var ready bool
        must(t, s.DB.QueryRow("SELECT ready_at<=clock_timestamp() FROM workflows WHERE id=$1", job.ID).Scan(&ready))
        return ready
    })
    must(t, worker.RunOne(context.Background()))
    w := state(t, s, job.ID)
    if w.State != "succeeded" || w.Attempt != 2 || effects(t, s) != 1 {
        t.Fatal(w)
    }
    observe(t, map[string]any{"backoff_ms": 250, "timer": "elapsed PostgreSQL clock", "attempts": w.Attempt, "post_calls": server.Calls.Load(), "effects": effects(t, s)})
}
func TestTrusted429ExhaustsPersistentBudget(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, _, worker := fixture(t, s, true)
    worker.Backoff = 0
    server.RejectRemaining.Store(20)
    for i := 0; i < 8; i++ {
        must(t, worker.RunOne(context.Background()))
    }
    w := state(t, s, job.ID)
    if w.State != "failed" || w.Attempt != 3 || w.MaxAttempts != 3 || server.Calls.Load() != 3 || effects(t, s) != 0 {
        t.Fatal(w)
    }
    observe(t, map[string]any{"state": w.State, "persistent_budget": w.MaxAttempts, "attempts": w.Attempt, "post_calls": server.Calls.Load(), "effects": effects(t, s)})
}
func TestUnknown404UsesOriginalKeyAndStopsAtBudget(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    _, _, worker := fixture(t, s, true)
    key := job.EffectKey()
    // An authoritative empty lookup is exercised over HTTP on each lost claim.
    for i := 1; i <= 3; i++ {
        claim, err := s.Claim(context.Background(), "lost", time.Second)
        must(t, err)
        if claim == nil {
            t.Fatal("missing claim")
        }
        must(t, s.Unknown(context.Background(), *claim))
        err = worker.Reconcile(context.Background(), job.ID)
        if i < 3 {
            must(t, err)
            if state(t, s, job.ID).EffectKey() != key {
                t.Fatal("key changed")
            }
        } else if !errors.Is(err, ErrManual) {
            t.Fatal("budget reset by lookup", err)
        }
    }
    for i := 0; i < 4; i++ {
        must(t, worker.RunOne(context.Background()))
    }
    w := state(t, s, job.ID)
    if w.State != "unknown" || w.Attempt != 3 || effects(t, s) != 0 {
        t.Fatal(w)
    }
    observe(t, map[string]any{"lookup": "HTTP 404", "original_key_preserved": w.EffectKey() == key, "attempts": w.Attempt, "persistent_budget": w.MaxAttempts, "state": w.State, "effects": effects(t, s)})
}
func TestPersistedVersionTwoRejectedByOldWorker(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, client, _ := fixture(t, s, true)
    _, err := s.DB.Exec("UPDATE workflows SET plan_version=2,plan=jsonb_set(plan,'{version}','2') WHERE id=$1", job.ID)
    must(t, err)
    p := child(t, s, client, "old-worker")
    p.finish(t)
    w := state(t, s, job.ID)
    if w.State != "incompatible" || w.Attempt != 0 || server.Calls.Load() != 0 {
        t.Fatal("unsupported version executed")
    }
    observe(t, map[string]any{"persisted_version": 2, "worker_version": 1, "state": w.State, "post_calls": server.Calls.Load()})
}
func TestTerminalTransactionRollsBackButEffectCanBeReconciled(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    server, _, worker := fixture(t, s, true)
    _, err := s.DB.Exec("ALTER TABLE events ADD CONSTRAINT reject_completed CHECK (kind <> 'completed')")
    must(t, err)
    if err = worker.RunOne(context.Background()); err == nil {
        t.Fatal("injected terminal transaction accepted")
    }
    var completed int
    must(t, s.DB.QueryRow("SELECT count(*) FROM events WHERE kind='completed'").Scan(&completed))
    before := state(t, s, job.ID)
    if before.State != "running" || before.Result != nil || completed != 0 || effects(t, s) != 1 {
        t.Fatal("terminal update not atomic")
    }
    _, err = s.DB.Exec("ALTER TABLE events DROP CONSTRAINT reject_completed")
    must(t, err)
    expired(t, s, job.ID)
    _, err = s.RecoverExpired(context.Background())
    must(t, err)
    must(t, worker.Reconcile(context.Background(), job.ID))
    after := state(t, s, job.ID)
    if after.State != "succeeded" || effects(t, s) != 1 || server.Calls.Load() != 1 {
        t.Fatal("reconciliation replayed effect")
    }
    observe(t, map[string]any{"fault": "events CHECK rejects completed", "after_rollback": before.State, "result_rolled_back": before.Result == nil, "completed_events": completed, "lease_expiry": "SQL-injected", "after_reconcile": after.State, "effects": effects(t, s), "post_calls": server.Calls.Load()})
}
func TestReceiptFingerprintMismatchCannotCompleteOrReconcile(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    _, client, _ := fixture(t, s, true)
    claim, err := s.Claim(context.Background(), "owner", time.Second)
    must(t, err)
    receipt, err := client.Call(context.Background(), *claim)
    must(t, err)
    receipt.Fingerprint = "wrong"
    if s.Complete(context.Background(), *claim, receipt) == nil {
        t.Fatal("mismatched receipt completed")
    }
    must(t, s.Unknown(context.Background(), *claim))
    unknown := state(t, s, job.ID)
    if s.Reconciled(context.Background(), unknown, &receipt, false) == nil {
        t.Fatal("mismatched receipt reconciled")
    }
    if state(t, s, job.ID).State != "unknown" || effects(t, s) != 1 {
        t.Fatal("invalid proof changed state")
    }
    observe(t, map[string]any{"completion_rejected": true, "reconciliation_rejected": true, "state": "unknown", "effects": effects(t, s)})
}
func TestEventsExcludeRawSensitiveInputs(t *testing.T) {
    s := db(t)
    ctx := context.Background()
    p := sample(true)
    p.Account = "private-account-DO-NOT-LOG-81925"
    job, err := s.Create(ctx, proposer, ID(), p)
    must(t, err)
    must(t, s.Approve(ctx, approver, job.ID, job.Fingerprint, time.Minute))
    _, _, worker := fixture(t, s, true)
    must(t, worker.RunOne(ctx))
    rows, err := s.DB.Query("SELECT kind,detail FROM events ORDER BY seq")
    must(t, err)
    defer rows.Close()
    n := 0
    for rows.Next() {
        var kind, detail string
        must(t, rows.Scan(&kind, &detail))
        for _, secret := range []string{p.Account, DevelopmentToolToken, "amount_minor", "Authorization", "book-lab-only"} {
            if strings.Contains(kind+detail, secret) {
                t.Fatalf("raw sensitive event content: %s", secret)
            }
        }
        n++
    }
    must(t, rows.Err())
    if n < 4 {
        t.Fatal("missing audit events")
    }
    observe(t, map[string]any{"events_checked": n, "raw_sensitive_markers_found": 0, "scope": "events kind/detail; workflow and tool ledger intentionally retain business fields"})
}
func TestCancellationDuringLookupCannotRequeue(t *testing.T) {
    s := db(t)
    job := approved(t, s, true)
    ctx := context.Background()
    claim, err := s.Claim(ctx, "owner", time.Second)
    must(t, err)
    must(t, s.Unknown(ctx, *claim))
    // The lookup handler authenticates its request, then cancellation wins before 404 returns.
    httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if r.Header.Get("Authorization") != "Bearer "+DevelopmentToolToken {
            t.Error("missing auth")
        }
        must(t, s.Cancel(ctx, proposer, job.ID))
        http.NotFound(w, r)
    }))
    defer httpServer.Close()
    client, err := NewToolClient(httpServer.URL)
    must(t, err)
    defer client.Close()
    worker := &Worker{Store: s, Tool: client}
    if !errors.Is(worker.Reconcile(ctx, job.ID), ErrManual) {
        t.Fatal("snapshot cancellation race allowed requeue")
    }
    w := state(t, s, job.ID)
    if w.State != "unknown" || !w.CancelRequested {
        t.Fatal(w)
    }
    observe(t, map[string]any{"cancel_during": "HTTP lookup", "lookup": "404", "state": w.State, "cancel_requested": w.CancelRequested, "effects": effects(t, s)})
}

func TestLoopbackBoundariesRejectRemoteTargets(t *testing.T) {
    for _, dsn := range []string{
        "postgres://book:book-lab-only@192.0.2.1:25432/book?sslmode=disable",
        "host=127.0.0.1,192.0.2.1 port=25432,25432 user=book password=book-lab-only dbname=book sslmode=disable",
    } {
        t.Setenv("BOOK_LAB_DSN", dsn)
        if s, err := Open(context.Background(), "wf_"+ID(), false); err == nil {
            _ = s.Close()
            t.Fatal("remote database target accepted")
        }
    }
    for _, endpoint := range []string{"http://192.0.2.1:8080", "https://127.0.0.1:8080", "http://user:secret@127.0.0.1:8080", "http://127.0.0.1:8080?token=secret"} {
        if c, err := NewToolClient(endpoint); err == nil {
            c.Close()
            t.Fatal("unsafe tool endpoint accepted")
        }
    }
    observe(t, map[string]any{"database_primary_and_fallback_restricted": true, "tool_endpoint_restricted": true, "localhost_resolution_pinned": true})
}

package lab

import (
    "bytes"
    "errors"
)

// Proposal is untrusted model content. It has no identity, approval, endpoint,
// retry, capability or executable credential fields.
type Proposal struct {
    Version     int    `json:"version"`
    Action      string `json:"action"`
    Account     string `json:"account"`
    AmountMinor int64  `json:"amount_minor"`
    Currency    string `json:"currency"`
}
type ToolContract struct {
    Name                             string
    Idempotency, AuthoritativeLookup bool
}

func AdmitProposal(raw []byte, contract ToolContract) (Plan, error) {
    if len(raw) > 4096 {
        return Plan{}, errors.New("proposal exceeds admission bound")
    }
    if contract.Name != "credit-service" || contract.Idempotency && !contract.AuthoritativeLookup {
        return Plan{}, errors.New("untrusted or incomplete tool capability contract")
    }
    var proposal Proposal
    if err := strictJSON(bytes.NewReader(raw), &proposal); err != nil {
        return Plan{}, err
    }
    plan := Plan{Version: proposal.Version, Action: proposal.Action, Account: proposal.Account, AmountMinor: proposal.AmountMinor, Currency: proposal.Currency, Idempotent: contract.Idempotency}
    return plan, plan.Validate()
}

CREATE TABLE workflows (
 id text PRIMARY KEY,
 tenant text NOT NULL,
 operation text NOT NULL,
 plan jsonb NOT NULL,
 fingerprint text NOT NULL,
 plan_version integer NOT NULL,
 state text NOT NULL CHECK (state IN ('waiting_approval','ready','running','unknown','succeeded','cancelled','failed','incompatible')),
 approved_by text NOT NULL DEFAULT '',
 approval_expires timestamptz,
 owner text NOT NULL DEFAULT '',
 fence bigint NOT NULL DEFAULT 0,
 attempt integer NOT NULL DEFAULT 0 CHECK (attempt>=0),
 max_attempts integer NOT NULL DEFAULT 3 CHECK (max_attempts BETWEEN 1 AND 10),
 lease_until timestamptz,
 ready_at timestamptz NOT NULL DEFAULT clock_timestamp(),
 cancel_requested boolean NOT NULL DEFAULT false,
 result jsonb,
 created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
 updated_at timestamptz NOT NULL DEFAULT clock_timestamp(),
 UNIQUE (tenant, operation)
);
CREATE INDEX ready_work ON workflows (ready_at, created_at) WHERE state='ready';
CREATE TABLE events (
 seq bigserial PRIMARY KEY,
 workflow_id text NOT NULL REFERENCES workflows(id),
 kind text NOT NULL,
 fence bigint NOT NULL,
 detail text NOT NULL DEFAULT '',
 created_at timestamptz NOT NULL DEFAULT clock_timestamp()
);
CREATE TABLE tool_effects (
 seq bigserial PRIMARY KEY,
 effect_key text NOT NULL UNIQUE,
 fingerprint text NOT NULL,
 account text NOT NULL,
 amount_minor bigint NOT NULL CHECK (amount_minor>0),
 receipt text NOT NULL,
 created_at timestamptz NOT NULL DEFAULT clock_timestamp()
);