--- title: "Flow and Ragjobs" description: "Abstract flow and ragjobs solve two execution problems that often appear together but require different semantics. flow is an in-process executor for large collections of semantically identified reque…" doc_version: 1 last_updated: 2026-10-17 canonical_url: "http://parc.yolo.scapegoat.dev/note/transcripts/2026/08/09/simplifying-ragkit-and-ragopt/flow_and_ragjobs_engineer_guide" --- # Flow and Ragjobs Abstract flow and ragjobs solve two execution problems that often appear together but require different semantics. flow is an in-process executor for large collections of semantically identified reque… ## Content # Abstract `flow` and `ragjobs` solve two execution problems that often appear together but require different semantics. `flow` is an in-process executor for large collections of semantically identified requests. It adds batching, bounded parallelism, duplicate coalescing, cache reuse, per-item retry, admission control, stable result ordering, and operational observation around an aligned batch processor. `ragjobs` is a durable orchestration kernel for coarse artifact-producing work. It persists canonical plans and run state, enforces dependency readiness, grants workers time-bounded fenced leases, recovers expired attempts, applies retry and budget policy, reuses semantically equivalent completed work, and publishes immutable artifacts through a deliberately small compare-and-swap boundary. The modules are designed to compose: a durable `ragjobs` handler can use `flow` to perform thousands of fine-grained provider calls, then commit one independently meaningful artifact. They are intentionally not one framework. `flow` does not persist a workflow graph; `ragjobs` does not understand embeddings, chunks, indexes, or other RAG-specific values. This guide first explains what the systems are, when to use each one, and how to run and extend them. It then develops the underlying principles: semantic identity, canonicalization, caches as partial functions, immutable artifacts, checked union, deterministic views, plan algebra, leases as capabilities, fencing, state-machine safety, and the distinction between denotation and operational trace. # Preface ## Audience This text is written for software engineers comfortable with Go, contexts, concurrency, hashing, transactions, and distributed-systems failure modes. The mathematical material is presented as a way to state API obligations precisely, not as an independent category-theory tutorial. ## Implementation status The guide describes the supplied standalone implementations, not an aspirational API. Both modules target Go 1.23 and use only the standard library. The executable validation performed for this edition includes: - `go test ./...` for both modules; - the `flow` embedding demonstration; - the `ragjobs` indexing demonstration; - the in-memory and file Store conformance tests included in the source. The in-memory `ragjobs` Store is the executable reference state machine. The file Store is a restartable single-process interpreter. PostgreSQL schema and transition contracts are included, but a live PostgreSQL adapter and compiled River adapter are not claimed by this implementation. ## Suggested reading paths - To use the libraries immediately, read Chapters 1 through 5. - To implement a new `flow` processor or cache, continue with Chapters 8 through 12. - To implement a durable `ragjobs` Store or transport, focus on Chapters 13 through 22. - To understand the mathematics and API design method, read Chapters 6, 7, 11, 16, 17, 25, and 26. # 1. System overview ## 1.1 Two execution scopes The fastest way to understand the split is by the lifetime of custody. `flow` owns work for one call to `flow.Run`. It assumes the process and call stack are the current custodian. It can retry a failed item and preserve deterministic result order, but it does not decide what happens after the process disappears. `ragjobs` owns work across process lifetimes. A plan and run survive worker replacement, deployment, duplicate delivery, and recovery. The Store, not an individual worker, is authoritative for current state. ![The two execution layers. `ragjobs` owns durable coarse-grained orchestration. `flow` owns fine-grained execution inside one handler attempt.](figures/layers.png){width=94%} A typical indexing execution is therefore hierarchical: ```text ragjobs run: build and publish index revision 1042 | +-- snapshot source +-- normalize corpus +-- chunk corpus +-- build lexical index +-- build dense index | | | +-- flow run: embed 30,000 chunk representations | +-- cache lookups | +-- provider batches | +-- per-item retries | +-- stable output collection +-- assemble bundle +-- verify +-- evaluate +-- gate +-- publish ``` ## 1.2 Responsibility matrix | Concern | `flow` | `ragjobs` | |---|---:|---:| | Typed request input and result | Yes | Handler-defined artifact ports | | Batch formation | Yes | No | | In-process bounded parallelism | Yes | Worker-level only | | Per-item retry | Yes | No | | Durable node retry | No | Yes | | Cache by semantic request identity | Yes | Yes, at node output level | | Restore caller order | Yes | Not generally relevant | | Persist a dependency graph | No | Yes | | Survive process death | No | Yes | | Leases and fencing | No | Yes | | Immutable artifact lineage | Result-specific | Core contract | | Source triggers and revisions | No | Extension/core run input | | Quality gate and publication | No | Durable stages/extensions | | Queue or transport adapter | No | Yes | ## 1.3 A durable job boundary A computation is a good durable node when its successful output is independently meaningful and worth recovering after process loss. Good examples: - source snapshot; - normalized corpus; - chunk-set manifest; - lexical index; - vector index; - assembled bundle; - verification report; - evaluation aggregate; - gate decision; - publication record. Usually poor examples: - normalize one chunk; - embed one chunk; - score one retrieval candidate; - evaluate one ordinary case; - parse one provider response. Those fine-grained operations usually belong inside `flow` or ordinary Go code. Durable nodes should be coarse enough that the control plane remains small, but fine enough that their outputs are reusable and their failures are operationally distinct. ## 1.4 What neither package is Neither module is a universal application framework. `flow` is not a workflow DSL, DAG scheduler, service container, actor runtime, or persistence layer. `ragjobs` is not a RAG library, index format, evaluation methodology, object store, or promise of arbitrary exactly-once execution. Pure transformations remain ordinary Go functions. The libraries appear at boundaries where execution policy or durable custody is actually required. # 2. Getting started with `flow` ## 2.1 Install and run the example The standalone distribution is a Go module with import path: ```text github.com/go-go-golems/flow ``` After extracting the module: ```bash cd composable-execution/flow go test ./... go run ./examples/embedding ``` The example submits 30 embedding-like calls, executes them in batches with four workers, deliberately fails one item on its first attempt, retries only that item, and returns: ```text calls=30 unique=30 batches=5 attempts=31 failed=0 ``` ## 2.2 The four kernel values The practical API is centered on four concepts: 1. `Key`: caller-defined semantic identity of one request. 2. `Call[I]`: a `Key` and typed local input. 3. `Processor[I,O]`: an aligned batch operation. 4. `Run`: the interpreter that applies execution mechanics. The package intentionally has no generic `Step` object and no pipeline graph. ## 2.3 Build complete semantic keys Consider embedding requests: ```go type EmbeddingRequest struct { Model string `json:"model"` Dimensions int `json:"dimensions"` Normalize bool `json:"normalize"` Text string `json:"text"` } type Vector []float64 ``` Construct a key before execution: ```go input := EmbeddingRequest{ Model: "embedding-model-v3", Dimensions: 1536, Normalize: true, Text: "How do I reset my password?", } key, err := flow.JSONKey("embedding", "v1", input) if err != nil { return err } call := flow.Call[EmbeddingRequest]{ Key: key, Input: input, } ``` The key must include every result-affecting input. Model revision, dimensions, normalization, representation text, response mode, prompt, or decoding policy belong in identity when they can change a successful value. Worker count, batch size, retry schedule, rate limiter, timestamps, metrics callbacks, and output paths do not belong in the request key when they only affect execution. A useful test is: > Can a cached successful value for request A be substituted for request B without changing the promised semantic result? If the answer is not unambiguously yes, the two requests need different keys. ## 2.4 Implement an aligned batch processor A processor consumes a batch and must return exactly one `ItemResult` for every input when its top-level error is nil: ```go processor := flow.ProcessorFunc[EmbeddingRequest, Vector]( func( ctx context.Context, batch []EmbeddingRequest, ) ([]flow.ItemResult[Vector], error) { providerResults, err := provider.EmbedBatch(ctx, batch) if err != nil { // The entire physical batch failed. return nil, flow.Transient(err) } results := make([]flow.ItemResult[Vector], len(batch)) for i := range batch { if providerResults[i].Err != nil { results[i].Err = classifyProviderError( providerResults[i].Err, ) continue } results[i].Value = providerResults[i].Vector } return results, nil }, ) ``` There are two failure levels: - a non-nil top-level `error` means the physical batch failed as a whole; - `ItemResult.Err` represents failure of one aligned item while others may succeed. If a service has no batch API, adapt an ordinary function with `flow.Each`: ```go processor := flow.Each( func(ctx context.Context, input EmbeddingRequest) (Vector, error) { return provider.EmbedOne(ctx, input) }, ) ``` `Each` is only an adapter. It does not create a workflow abstraction. ## 2.5 Execute calls ```go outcomes, report, err := flow.Run( ctx, flow.Executor{ Cache: embeddingCache, }, calls, processor, flow.Config[EmbeddingRequest, Vector]{ Workers: 8, BatchSize: 128, Retry: flow.RetryPolicy{ MaxAttempts: 4, Backoff: flow.Backoff{ Base: 250 * time.Millisecond, Cap: 5 * time.Second, Multiplier: 2, }, }, Codec: flow.JSONCodec[Vector]{}, }, ) if err != nil { return err } ``` `Run` returns two different forms of failure: - executor or custody failures are returned as the third result; - domain/item failures remain in aligned `Outcome.Err` values. When every item must succeed: ```go vectors, err := flow.Values(outcomes) if err != nil { return err } ``` When partial success is acceptable, inspect each outcome directly: ```go for i, outcome := range outcomes { if outcome.Err != nil { log.Printf("call %d failed after %d attempts: %v", i, outcome.Attempts, outcome.Err) continue } consume(outcome.Value) } ``` ## 2.6 What `Run` actually does Suppose the input keys are: ```text A B C A D ``` and `A` is cached. `flow.Run` behaves conceptually as follows: ```text caller order: A B C A D unique requests: A B C D cache hits: A first execution: B C D temporary failure: D retry wave: D expanded outcomes: A B C A D ``` Calls with the same key are executed once and receive the same outcome. Successful cache misses are persisted immediately. A later retry wave contains only failed items. Final outcomes are expanded back to the caller's original order. ![The `flow` interpretation: group by semantic identity, reuse successes, execute retry waves, and restore caller order.](figures/flow-semantics.png){width=94%} ## 2.7 Caching A cache is intentionally stricter than a mutable key/value store: ```go type Cache interface { Get(context.Context, flow.Key) (flow.CacheEntry, bool, error) Put(context.Context, flow.CacheEntry) error } ``` `Put` must accept either: - a new key and value; - the exact same value already stored for the key. It must reject a different value for an existing key with `flow.ErrCacheConflict`. That conflict is evidence of an incomplete key, nondeterministic output under a deterministic contract, schema drift, or corruption. It is not an eviction or last-write-wins case. When a cache is configured, a `Codec[O]` is required because cache entries store validated bytes. The module includes memory and atomic file caches. ## 2.8 Retry classification Unknown errors fail closed as permanent. Retry requires explicit classification: ```go return flow.Transient(err) ``` or: ```go return flow.RateLimited(2*time.Second, err) ``` The default classifier retries only transient and rate-limited failures. A custom classifier can map provider-specific errors without changing the executor. An important property is that a successful item from an earlier wave is not rerun merely because another item failed. ## 2.9 Admission and resource policy Admissions run before each physical batch attempt: ```go type Admitter interface { Admit(context.Context, flow.Work) error } ``` `Work` records keys, attempt number, item count, and caller-defined units. Included implementations provide finite budgets and a local token bucket. ```go budget, err := admission.NewBudget( 1_000_000, admission.Units, ) if err != nil { return err } limiter, err := admission.NewTokenBucket( 20, // units per second 40, // burst admission.Calls, ) if err != nil { return err } executor := flow.Executor{ Cache: embeddingCache, Admissions: []flow.Admitter{budget, limiter}, } ``` Every admitted attempt is charged, including attempts that later fail. This models actual provider or compute consumption rather than only successful output. ## 2.10 Observation Observers receive bounded operational events: ```go executor.Observer = flow.ObserverFunc(func(event flow.Event) { metrics.Record(event) }) ``` Observers cannot veto or mutate execution. Event order belongs to the operational trace, not semantic output identity. ## 2.11 When to use `flow` Use `flow` when the unit of work is a large collection of independent or batchable semantic requests inside one process attempt. Typical uses include: - embeddings; - generation or classification batches; - reranking calls; - generated document representations; - document conversion; - evaluation cases within a durable shard; - provider-backed enrichment. Do not use `flow` to represent a multi-hour production build, publication lifecycle, or database change stream. Those require durable custody. # 3. Getting started with `ragjobs` ## 3.1 Install and run the example The module import path is: ```text github.com/go-go-golems/ragjobs ``` After extraction: ```bash cd composable-execution/ragjobs go test ./... go run ./examples/indexing ``` The indexing example executes an 11-node plan. One dense-stage attempt is deliberately rate limited, so the completed run has 12 attempts and 38 durable events. ## 3.2 The durable kernel The main values and interfaces are: - `Ref` and `Artifact`: immutable semantic outputs; - `Plan` and `NodeSpec`: canonical versioned DAG data; - `RunInput`: immutable source and configuration input; - `Store`: the authoritative durable state machine; - `Handler` and `Registry`: application-defined node protocols; - `Worker`: a transport-neutral lease interpreter; - `Lease`: a time-bounded fenced capability. ![A reference indexing plan. Independent lexical and dense branches converge on one immutable bundle and publication path.](figures/ragjobs-dag.png){width=94%} ## 3.3 Define a plan A minimal one-node plan is: ```go plan, err := ragjobs.Finalize(ragjobs.Plan{ Name: "example-build", Version: "v1", Nodes: []ragjobs.NodeSpec{{ ID: "build", Handler: ragjobs.HandlerRef{ Kind: "example.build", Version: "v1", }, Args: json.RawMessage(`{"format":"compact"}`), Queue: "cpu", Outputs: []ragjobs.Port{{ Name: "result", Type: "example/result", }}, Retry: ragjobs.RetryPolicy{ MaxAttempts: 3, }, Cache: ragjobs.CacheContent, Failure: ragjobs.FailRun, MaxCost: 10, }}, }) if err != nil { return err } ``` `Finalize`: 1. clones caller-owned data; 2. applies defaults; 3. canonicalizes JSON arguments; 4. sorts dependencies, ports, labels, and nodes where order is irrelevant; 5. validates identifiers, dependencies, ports, and acyclicity; 6. computes a stable plan identity. A persisted plan refers to a handler by `(kind, version)`. If handler semantics change observably, the protocol version must change or the new semantics must be represented in canonical arguments or run values. ## 3.4 Model dependencies and artifact ports A realistic plan includes direct dependencies: ```go ragjobs.NodeSpec{ ID: "assemble-bundle", Handler: ragjobs.HandlerRef{ Kind: "rag.assemble-bundle", Version: "v2", }, Queue: "cpu", DependsOn: []string{ "build-lexical", "embed-dense", }, Inputs: []ragjobs.Port{ {Name: "lexical", Type: "rag/lexical-index"}, {Name: "vectors", Type: "rag/vector-index"}, }, Outputs: []ragjobs.Port{ {Name: "bundle", Type: "rag/index-bundle"}, }, } ``` Ports are nominal compatibility checks. The current runtime verifies that every declared input type is provided by a direct dependency. Handlers and artifact readers remain responsible for structural validation of actual bytes. Plans can be written directly as data. Optional `Atom`, `Then`, and `Parallel` helpers are available for reusable plan constructors and algebraic tests. They do not replace the persisted canonical `Plan`. ## 3.5 Register handlers A handler implements one versioned application protocol: ```go registry := ragjobs.NewRegistry() err = registry.Register( ragjobs.HandlerRef{ Kind: "example.build", Version: "v1", }, ragjobs.HandlerFunc(func( ctx context.Context, invocation ragjobs.Invocation, ) (ragjobs.Result, error) { ref, err := ragjobs.JSONRef( "example/result", "v1", map[string]any{ "cursor": invocation.Input.Trigger.Cursor, "format": "compact", }, ) if err != nil { return ragjobs.Result{}, err } return ragjobs.Result{ Outputs: map[string]ragjobs.Artifact{ "result": { Ref: ref, MediaType: "application/json", }, }, Cost: 1, }, nil }), ) if err != nil { return err } ``` Handlers receive immutable run input, canonical node specification, direct dependency artifacts, semantic invocation key, attempt number, fence, and worker identity. They return semantic outputs and measured cost. They do not mutate run or node state directly. ## 3.6 Create a run ```go store := memory.New() created, err := store.CreateRun( ctx, ragjobs.CreateRunRequest{ Plan: plan, Input: ragjobs.RunInput{ Trigger: ragjobs.Trigger{ Kind: "manual", Source: "example/source", Cursor: "revision-1", Payload: json.RawMessage(`{}`), }, Values: map[string]json.RawMessage{ "configuration": json.RawMessage( `{"mode":"production"}`, ), }, Budget: ragjobs.Budget{ MaxAttempts: 10, MaxCost: 100, }, }, }, ) if err != nil { return err } ``` The trigger separates semantic source position from operational metadata: - `Kind`, `Source`, `Cursor`, `Payload`, and `Values` participate in semantic input identity; - `DedupKey`, `OccurredAt`, labels, and execution budget do not. A deduplication key suppresses accidental duplicate run creation. It is not a semantic result identity. ## 3.7 Run workers ```go worker := ragjobs.Worker{ Store: store, Registry: registry, ID: "worker-1", Queues: []string{"cpu"}, LeaseDuration: 2 * time.Minute, HeartbeatInterval: 40 * time.Second, Concurrency: 4, } ``` For a long-running process: ```go if err := worker.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { return err } ``` For tests and local programs, call `RunOne` in a loop: ```go for { run, err := store.GetRun(ctx, created.RunID) if err != nil { return err } if run.State.Terminal() { break } worked, err := worker.RunOne(ctx) if err != nil { return err } if !worked { time.Sleep(10 * time.Millisecond) } } ``` ## 3.8 Inspect state and events ```go run, err := store.GetRun(ctx, created.RunID) if err != nil { return err } fmt.Println(run.State) fmt.Println(run.AttemptsUsed) fmt.Println(run.CostConsumed) for nodeID, node := range run.Nodes { fmt.Println(nodeID, node.State, node.OutputDigest) } ``` Events are read by sequence number: ```go events, err := store.Events( ctx, created.RunID, 0, // after sequence 100, // limit ) ``` The event log is durable audit history. Current run and node rows are materialized state for efficient operation; the event stream is not the only source of truth. ## 3.9 Memory, file, and production Stores The supplied interpreters are: - `store/memory`: serializes every transition under one mutex and serves as the executable reference semantics; - `store/file`: stores the same reference state using atomic file snapshots and is suitable for restartable single-process use. A production multi-process Store should implement the same interface transactionally. The included `storetest` package supplies reusable conformance checks. PostgreSQL migrations document the intended control-plane schema and fenced transition predicates. ## 3.10 Artifacts A `Ref` is semantic identity: ```go type Ref struct { Kind string Version string Digest string } ``` An `Artifact` adds custody metadata: ```go type Artifact struct { Ref Ref MediaType string Size int64 Location string Metadata map[string]string } ``` Location and metadata do not define semantic equality. Two artifact descriptors can refer to the same semantic value stored at different locations. The `artifactfs` package provides immutable content-addressed local storage. It writes to temporary storage, computes the `Ref`, publishes using create-if-absent semantics, and verifies the resulting bytes. ## 3.11 Failure classification Handlers explicitly classify failures: ```go return ragjobs.Result{}, ragjobs.Transient(err) ``` ```go return ragjobs.Result{}, ragjobs.RateLimited( retryAfter, err, ) ``` ```go return ragjobs.Result{}, ragjobs.Conflict(err) ``` ```go return ragjobs.Result{}, ragjobs.Permanent(err) ``` Transient, rate-limited, conflict, and lease-expired failures are retryable subject to node policy and run budgets. Unknown errors fail closed as permanent. `FailRun` cancels remaining nonterminal work after terminal node failure. `ContinueRun` allows causally independent branches to finish, while descendants of failed dependencies become skipped. ## 3.12 Leases and fencing A claim grants a worker a lease: ```text (run, node, attempt, fence, worker, expiration) ``` The lease is a capability, not just metadata. Every heartbeat, completion, and failure submission must present the current lease fields. If Worker A owns fence 7, stops heartbeating, and Worker B later claims fence 8, any completion from fence 7 is rejected. This is stale-worker exclusion. ![The durable run/node state machine. A retry or lease recovery returns work to availability; every new claim increments the fence.](figures/state-machine.png){width=94%} Fencing protects the `Store`. External mutable systems must also use a fence, idempotency key, compare-and-swap, or reconciliation protocol. A database fence cannot undo an already committed provider side effect. ## 3.13 Semantic result reuse When `NodeSpec.Cache` is `CacheContent`, the worker looks up the node's semantic invocation key before running the handler. That key binds: - handler kind and version; - canonical node arguments; - semantic run input; - direct dependency output digests. Attempts, worker IDs, leases, timing, labels, and budgets are excluded. Like the `flow` cache, the durable cache is functional: one semantic key cannot map to two different output digests. A conflict is a correctness incident. ## 3.14 Publication Expensive work produces immutable candidate artifacts. Publication changes a small mutable alias: ```text demo/knowledge/production -> rag/index-bundle@v1:sha256... ``` The `publish` package implements compare-and-swap using expected generation and optional expected digest. Ordered source revisions are monotone, except through a separate privileged rollback procedure. An ambiguous retry of the exact same target and revision is idempotent. A late older revision is rejected. A concurrent unrelated publisher causes a conflict rather than unconditional overwrite. ![Publication changes one alias atomically after immutable work has completed and been verified.](figures/publication.png){width=91%} ## 3.15 Source changes and triggers The `trigger` package normalizes source-ordered changes, rejects conflicting duplicate event IDs, sorts them deterministically, and selects only a contiguous prefix after a committed revision. If revisions 11 and 13 are present but 12 is missing, revision 13 is not adopted. This prevents a delayed source event from being silently skipped. For a source database separate from the control database, use a transactional outbox at the source, an at-least-once relay into a deduplicated inbox, contiguous coalescing, and a durable run created from the resulting batch. ![Source-side outbox, at-least-once relay, contiguous adoption, durable build, and monotone publication.](figures/outbox.png){width=92%} ## 3.16 When to use `ragjobs` Use `ragjobs` when work must be explainable and recoverable beyond one process lifetime: - production indexing and publication; - change-driven rebuilds; - long evaluation suites; - optimization trials; - artifact verification and promotion; - durable backfills; - sharded fan-out/fan-in; - workflows with audit, budgets, and operator intervention. Do not use it merely to obtain goroutine concurrency or to persist every fine-grained provider request. # 4. Using `flow` inside `ragjobs` ## 4.1 The standard composition A durable handler can use `flow` internally: ```go func embedDenseHandler( flowExecutor flow.Executor, objects *artifactfs.Store, provider flow.Processor[EmbeddingRequest, Vector], ) ragjobs.Handler { return ragjobs.HandlerFunc(func( ctx context.Context, invocation ragjobs.Invocation, ) (ragjobs.Result, error) { chunks, err := loadChunkSet( ctx, invocation.Dependencies["chunk-corpus"]["chunks"], ) if err != nil { return ragjobs.Result{}, ragjobs.Permanent(err) } calls, err := makeEmbeddingCalls(chunks) if err != nil { return ragjobs.Result{}, ragjobs.Permanent(err) } outcomes, report, err := flow.Run( ctx, flowExecutor, calls, provider, flow.Config[EmbeddingRequest, Vector]{ Workers: 8, BatchSize: 128, Retry: flow.RetryPolicy{ MaxAttempts: 4, }, Codec: flow.JSONCodec[Vector]{}, }, ) if err != nil { return ragjobs.Result{}, ragjobs.Transient(err) } vectors, err := flow.Values(outcomes) if err != nil { return ragjobs.Result{}, mapFlowFailure(err) } artifact, err := writeVectorIndex( ctx, objects, chunks, vectors, ) if err != nil { return ragjobs.Result{}, ragjobs.Transient(err) } return ragjobs.Result{ Outputs: map[string]ragjobs.Artifact{ "vectors": artifact, }, Cost: int64(report.Attempts), }, nil }) } ``` The durable node has one semantic artifact result. The inner executor handles thousands of identified calls. If the process dies before durable completion, `ragjobs` may run the handler again; `flow`'s semantic cache avoids repeating successful fine-grained work. ## 4.2 Identity hierarchy The identities are nested but distinct: ```text ragjobs plan ID identifies the canonical durable graph and policy ragjobs semantic invocation key identifies one node result under run input and dependency artifacts flow request key identifies one fine-grained provider or algorithm result attempt and fence identify one physical ownership interval ``` Changing `flow.Workers` does not change a request key. Changing a node's retry limit may change the plan identity but not the semantic invocation key. Reclaiming a node changes attempt and fence but not its semantic identity. ## 4.3 Nested retry discipline There are two valid retry scopes: - inner `flow` retry handles short-lived item or provider failures; - outer `ragjobs` retry handles process death, exhausted inner retry, whole-node failure, and recovery. The product of both attempt limits matters for latency and cost. Do not let two durable layers independently schedule the same physical retry. If River owns physical retry timing, the control plane must not also create an independent delayed attempt for the same dispatch. ## 4.4 Budget discipline `flow` admissions charge fine-grained attempts. `ragjobs` reserves and records node-level maximum and actual cost. A practical handler reports a conservative aggregate cost to the outer layer, while inner budgets prevent runaway provider consumption inside the attempt. The units do not have to be currency; they may represent tokens, requests, GPU seconds, or a stable internal accounting unit. ## 4.5 Trace linkage Use stable identifiers to connect traces: ```text ragjobs run ID ragjobs node ID attempt fence ragjobs semantic key flow request key provider request ID artifact Ref ``` Do not make timing, callback order, or trace IDs part of semantic identity. # 5. Boundaries, deployment, and operations ## 5.1 Decision table | Question | Prefer | |---|---| | Is this a large set of similar calls in one process? | `flow` | | Must work survive process or machine death? | `ragjobs` | | Is the output an independently meaningful immutable artifact? | Durable `ragjobs` node | | Is the unit one provider request or one record? | Usually `flow` | | Do dependencies and publication need durable audit? | `ragjobs` | | Do I only need ordinary function composition? | Plain Go | | Do I need a global top-k or reducer over a fixed set? | Plain Go/view after collection | | Do I need a queue transport? | `ragjobs` adapter, not `flow` | ## 5.2 Recommended indexing granularity A production RAG build often uses roughly 8 to 20 durable nodes: ```text snapshot -> extract -> chunk -> lexical || dense -> assemble -> verify -> evaluate shards -> aggregate -> gate -> publish -> cleanup ``` A dense node may execute tens of thousands of `flow` calls. An evaluation shard may execute hundreds of cases with `flow`. This keeps the durable control plane small without sacrificing fine-grained reuse. ## 5.3 Worker pools and queues Separate durable queues by resource class when operationally useful: - control and snapshots; - CPU indexing; - embedding/provider work; - evaluation; - publication; - low-priority backfills. Queue selection is operational policy in the plan. It does not define artifact semantics. ## 5.4 Immutable data plane, mutable control plane The recommended production split is: ```text mutable transactional control plane runs, nodes, attempts, leases, fences, events, cache index, trigger inbox, publication aliases immutable data plane source snapshots, corpora, chunks, vectors, indexes, bundles, evaluations, gate evidence ``` Expensive nodes never edit the active index in place. They build immutable candidates. Publication changes one alias after verification and policy. ## 5.5 Operational failure interpretation A few incidents have strong semantic meaning: **Semantic cache conflict.** Quarantine the relevant key or handler version. Investigate missing identity inputs, moving dependencies, nondeterminism, or corruption. Do not retry as an ordinary transient. **Repeated lease recovery.** Inspect heartbeat intervals, long non-interruptible calls, worker pauses, and database time. Recovery is functioning, but the execution profile is unhealthy. **Outbox gap.** Preserve later revisions and repair the missing source event or cursor protocol. Do not skip ahead to reduce backlog. **Gate rejection.** Leave the active alias unchanged. The run may have succeeded operationally while producing a valid rejection decision. **Publication conflict.** Resolve whether the active target is already the desired artifact, a newer source revision, or an unrelated concurrent candidate. Never fall back to unconditional replacement. ## 5.6 Extending the systems Extensions are packages implementing narrow interfaces, assembled explicitly by the application. For `flow`: - `Cache`; - `Codec`; - `Classifier`; - `Admitter`; - `Observer`; - `Clock`; - domain-specific `Processor`. For `ragjobs`: - `Store`; - `Handler`; - artifact stores; - publication stores; - source trigger adapters; - queue/transport adapters; - evaluators and gate handlers. There is no global plugin manager or implicit discovery mechanism. An application's registry and constructor code are the composition root. ![The principal graft points. Extensions implement narrow capabilities around the two small kernels.](figures/grafts.png){width=60%} ## 5.7 Current production boundary The supplied implementations are appropriate for: - local and test use through the in-memory Stores; - restartable single-process use through file Stores; - use as executable specifications for production adapters. Before automatic production publication, a PostgreSQL or River-backed deployment should add: - Store conformance tests against the in-memory implementation; - multi-process claim contention tests; - crash injection at every transition boundary; - database failover and authoritative-time tests; - object-store conditional publication tests; - source outbox duplicate, reorder, and gap tests; - application-specific artifact and evaluation equivalence tests. The remaining chapters explain why these APIs have their present shapes and state their mathematical and operational contracts in detail. # 6. Semantic foundations ## 6.1 Information, views, and traces Three data categories recur throughout both packages. ### Semantic information Semantic information is the value the computation means to establish. Examples include an embedding vector, an immutable chunk set, an index descriptor, a gate decision, or a published artifact reference. Its identity excludes worker IDs, timestamps, retry counters, queue latency, and callback order. ### View A view is a deterministic projection of fixed semantic information under a versioned policy. Ranking, top-k selection, context packing, report rendering, and a quality gate are views. Views may reorder or remove items. They are not generally monotone with respect to candidate-set growth. ### Trace A trace records execution history: cache hits, batch starts, attempts, workers, leases, heartbeats, completion order, delays, costs, and failures. Two runs may have equal semantic output and unequal traces. This is expected, not a defect. ## 6.2 Three identities The implementation separates three identity classes. ![Three identity classes: semantic value or request, durable plan, and physical attempt/trace.](figures/identities.png){width=88%} 1. **Semantic identity** answers: *May one successful value substitute for another?* 2. **Plan identity** answers: *Which durable graph and policy was created?* 3. **Attempt identity** answers: *Which physical ownership interval performed this work?* For example, changing `Workers: 8` to `Workers: 16` does not change an embedding request key. Changing the model revision does. Changing a durable node retry count changes the finalized plan, but does not necessarily change the semantic invocation key of the node. A recovered node gets a new attempt and fence, while retaining the same semantic key if its semantic input is unchanged. Conflating these identities is a primary source of invalid cache reuse and irreproducible artifacts. ## 6.3 Content-addressed references Both modules use domain-separated hashes. Abstractly, a semantic reference is: $$ \operatorname{Ref}(k,v,x) = \bigl(k,v,H(\operatorname{frame}(d,k,v,x))\bigr), $$ where: - $k$ is a kind; - $v$ is a schema or protocol version; - $x$ is a canonical byte representation; - $d$ is a package-specific domain separator; - `frame` length-prefixes fields, preventing concatenation ambiguity. Domain separation prevents identical bytes in two conceptual domains from sharing an accidental identity. Versioning prevents an altered canonicalization or semantic contract from silently reusing old identities. The hash is not a proof that two independently designed encoders mean the same thing. Correctness still depends on a documented equivalence relation for the canonical bytes. ## 6.4 Partial functions and semantic caches A semantic cache is modeled as a partial function: $$ M : K \rightharpoonup V. $$ For one key $k$, either no successful value is known, or exactly one successful value is known. A second unequal value for the same key is a contradiction: $$ M(k)=v_1 \land v_1 \neq v_2 \quad\Rightarrow\quad \operatorname{Put}(k,v_2)=\operatorname{Conflict}. $$ This is stricter than an ordinary mutable cache. "Last writer wins" would hide incomplete identity, nondeterminism, corruption, or schema drift. Cacheability is therefore a semantic assertion: > For every world that agrees on the complete semantic input, all successful observations admitted under this key are equivalent under the promised output equality. Operational traces and costs may differ between fresh execution and reuse. ## 6.5 Checked union and ACI laws Several states in the system accumulate keyed information. The ideal merge is checked union: ```text new key -> insert same key + same canonical value -> duplicate, accept same key + different value -> conflict ``` For compatible states, merge satisfies: $$ (a \sqcup b) \sqcup c = a \sqcup (b \sqcup c), $$ $$ a \sqcup b = b \sqcup a, $$ $$ a \sqcup a = a. $$ These are associativity, commutativity, and idempotence. They explain why duplicate result delivery, regrouped batches, and completion order can be semantically harmless when outputs are content addressed and conflicts fail closed. The packages do not force every object into one generic semilattice type. They apply the law locally where it buys a concrete property: cache entries, immutable artifacts, normalized changes, dependency outputs, and candidate accumulation in adapters. ## 6.6 Pure functions remain ordinary functions A pure deterministic value transformation belongs to ordinary Go: ```go normalized := Normalize(source) chunks := Chunk(normalized) manifest := Manifest(chunks) ``` No `Step` wrapper is needed merely to express composition. The framework appears only when one of the following is required: - effect execution; - batching or bounded concurrency; - cache custody; - resource admission; - durable ownership; - persisted dependency readiness; - immutable artifact custody; - atomic publication. This principle keeps the semantic center visible. # 7. A method for small extensible kernels ## 7.1 Begin with observation levels Before introducing an interface, state what an observer is allowed to distinguish. For `flow`, semantic equality ignores: - worker count; - batch boundaries; - completion order; - retry delay; - cache hit versus fresh execution; - observer callback order. For `ragjobs`, node semantic equality ignores: - run ID; - worker ID; - attempt number; - fence; - lease duration; - event occurrence time; - queue latency. The trace retains these distinctions. The semantic result does not. ## 7.2 Put laws before fields A field belongs in the kernel only when it participates in a stated invariant or transition. For example: - `flow.Key` exists because cache substitution and duplicate coalescing require semantic identity; - `ragjobs.Fence` exists because stale-worker exclusion requires a monotonically increasing ownership token; - `Artifact.Ref` exists because semantic output identity must not depend on storage location; - `ConcurrencyKey` exists because some incomparable DAG nodes still require serialization. Conversely, a display label should not be overloaded as a unique identifier simply because it is convenient. ## 7.3 Prefer algebraic values to manager objects A plan is data. A reference is data. A retry policy is data. A lease is data. A result is data. The interpreter is the component that owns behavior. This makes it possible to: - canonicalize and hash plans; - persist historical protocols; - test transitions independently of handlers; - provide multiple Store implementations; - replay or inspect values without starting workers. ## 7.4 Use narrow graft points A graft point is an interface where the kernel truly needs an external interpretation. The reference modules use small interfaces such as: ```go type Processor[I, O any] interface { Process(context.Context, []I) ([]ItemResult[O], error) } ``` ```go type Store interface { CreateRun(context.Context, CreateRunRequest) (CreateRunResult, error) Claim(context.Context, ClaimRequest) ([]Lease, error) Heartbeat(context.Context, Lease, time.Time, time.Duration) (Lease, error) Complete(context.Context, Completion) error Fail(context.Context, Failure) error RecoverExpired(context.Context, time.Time, int) (int, error) CancelRun(context.Context, string, string, time.Time) error LookupCached(context.Context, string) (CachedResult, bool, error) GetRun(context.Context, string) (RunSnapshot, error) Events(context.Context, string, int64, int) ([]Event, error) ListRuns(context.Context, ListFilter) ([]RunSnapshot, error) } ``` Each interface exists because alternate implementations are required. There is no global plugin lifecycle or hidden service locator. Applications assemble concrete implementations explicitly. ## 7.5 Keep optional algebra optional `ragjobs.Plan` can be built directly as data. `Atom`, `Then`, and `Parallel` are optional syntax for reusable constructors and algebraic tests. They do not become the only way to express a plan. This distinction matters. A small algebra can clarify laws without becoming a mandatory domain-specific language. # 8. `flow`: mandate and non-goals ## 8.1 Contract `flow` executes a finite slice of semantically identified typed requests inside one process-level call. It adds operational mechanics without claiming ownership of the larger application procedure. Its core use case is: > Execute many related requests with bounded concurrency, batching, content reuse, request-level retry, resource admission, stable output order, and bounded observation. The package does not persist a DAG, schedule work across machines, recover leases, publish artifacts, or determine which application stages exist. ## 8.2 Why the old `Step` shape was rejected A broad step object commonly begins as a typed function with a name. It then accumulates cache identity, workers, retry, batch policy, barriers, budgets, metering, callbacks, reporting, and composition metadata. Several problems follow. First, semantic and operational settings become adjacent fields, encouraging invalid identity rules. Second, a local batching helper begins to look like a workflow engine. Third, composition laws become ambiguous because semantic outputs, reports, callbacks, and timing are all observable through the same object. Fourth, simple functions require framework wrappers even when no execution service is needed. The new `flow` removes the central `Step[I,O]` concept. The actual operation is a `Processor[I,O]`; the actual request identity is a `Key`; the interpreter is `Run`; resource and custody behaviors are grafts on the interpreter. ## 8.3 Kernel inventory The user can understand the kernel by reading four files: | Concept | Role | |---|---| | `Key` | caller-owned semantic request identity | | `Call[I]` | semantic key plus local typed input | | `Processor[I,O]` | aligned batch effect | | `Run` | operational interpreter | Everything else is either a policy value, an extension interface, a concrete extension, or a convenience type. ![`flow` separates semantic request identity from operational batching, retry, cache, admission, and trace.](figures/flow-semantics.png){width=94%} # 9. `flow` semantic API ## 9.1 `Key` The key type is intentionally transparent: ```go type Key struct { Kind string `json:"kind"` Version string `json:"version"` Digest string `json:"digest"` } ``` Construction uses a domain-separated SHA-256 digest over a kind, version, and canonical payload. The caller, not the executor, owns the equivalence relation. ```go key, err := flow.JSONKey( "embedding/request", "v2", struct { Model string `json:"model"` ModelRevision string `json:"model_revision"` Representation string `json:"representation"` TextDigest string `json:"text_digest"` Dimensions int `json:"dimensions"` Normalize bool `json:"normalize"` }{ Model: "embed-x", ModelRevision: "2026-07-17", Representation: "raw", TextDigest: textDigest, Dimensions: 1536, Normalize: true, }, ) ``` A correct key contains every setting that may alter the successful value. Typical semantic fields include: - model and pinned model revision; - prompt and response schema digests; - input text or artifact identity; - output dimension; - normalization; - tool mode; - decoding and reasoning settings; - index snapshot and query policy; - adapter semantic version. Typical operational exclusions include: - worker count; - batch size; - retry count and delay; - rate limiter implementation; - trace observer; - output path; - wall-clock occurrence time. The safest key contract is: > Two calls share a `Key` exactly when `flow` is permitted to execute one successful result and substitute it for both. ## 9.2 Canonicalization and versioning `JSONKey` uses `encoding/json`. Go's encoder sorts map keys, but persistent protocols still require discipline around floats, omitted fields, custom marshaling, and schema evolution. When equivalence changes, change the key version. Do not silently reinterpret persisted keys. For highly stable protocols, define a specific canonical request struct rather than hashing a large runtime configuration object. This makes field classification reviewable and prevents secrets or operational settings from entering persisted identity accidentally. ## 9.3 `Call[I]` ```go type Call[I any] struct { Key Key Input I } ``` `Input` is the local typed value needed to invoke the processor. `Key` is the stable semantic identity. The two need not be the same representation. For example, `Input` may contain a provider client-ready byte slice or an in-memory object, while the key contains only canonical semantic fields. This avoids forcing generic serialization into the hot path. Calls with equal keys are coalesced inside one `Run`. They must also report equal admission units. If the caller constructs one key for semantically unequal inputs, the package cannot repair that error. It will execute one and return the same outcome to all duplicates, exactly as the key contract permits. ## 9.4 `Processor[I,O]` ```go type Processor[I, O any] interface { Process(context.Context, []I) ([]ItemResult[O], error) } ``` A processor consumes one physical batch and returns aligned per-item results. ```go type ItemResult[T any] struct { Value T Err error } ``` There are two failure levels: - a top-level processor error means the whole physical batch failed; - an `ItemResult.Err` means one item failed while others may have succeeded. This distinction is essential for partial retry. A provider batch endpoint can return vectors for 126 items and errors for 2. `flow` commits and caches the 126 successes immediately, then retries only the 2 failures. The alignment law is strict: $$ \lvert \operatorname{Process}(xs) \rvert = \lvert xs \rvert $$ whenever the top-level error is nil. Violating the law is an executor custody error, not an item failure. `ProcessorFunc` adapts a batch function. `Each` adapts an ordinary per-item function without adding composition semantics: ```go processor := flow.Each(func(ctx context.Context, input Input) (Output, error) { return compute(ctx, input) }) ``` ## 9.5 `Outcome` and error levels `Run` returns aligned outcomes: ```go type Outcome[T any] struct { Key Key Value T Err error Cache CacheStatus Attempts int } ``` Item failures are values in `Outcome.Err`. The top-level error from `Run` is reserved for custody or interpreter failure, such as: - invalid keys; - corrupt cache entries; - semantic cache conflicts; - failed cache persistence; - malformed processor alignment; - context cancellation affecting the interpreter; - invalid configuration. This separation prevents one ordinary failed request from erasing successful sibling results. # 10. `flow.Run` operational semantics ## 10.1 Signature ```go func Run[I, O any]( ctx context.Context, executor Executor, calls []Call[I], processor Processor[I, O], config Config[I, O], ) ([]Outcome[O], Report, error) ``` `Executor` contains run-scoped collaborators: ```go type Executor struct { Cache Cache Admissions []Admitter Observer Observer Clock Clock } ``` `Config` contains operational policy and output encoding: ```go type Config[I, O any] struct { Workers int BatchSize int Retry RetryPolicy Codec Codec[O] Units func(I) int } ``` No field in `Executor` or `Config` is automatically included in request identity. Semantic settings must already be represented by each call's `Key`. ## 10.2 Execution phases The interpreter implements the following phases. ### Phase 1: validate and group Every key is validated. Calls are grouped by key in first-occurrence order. The group stores all original positions so the final result can be expanded back to caller order. If $n$ calls contain $u$ unique keys, the executor performs at most $u$ semantic executions in the run. ### Phase 2: cache lookup For each unique key, the executor queries the optional cache. A hit must pass: - stored-key equality; - encoded-value digest validation; - output decoding. A corrupt entry is not treated as a miss. Failing closed avoids silently replacing a damaged reproducibility record with an unrelated fresh result. ### Phase 3: form a retry wave All pending unique items enter an attempt wave. Items are ordered by stable unique position and partitioned into batches of `BatchSize`. Batch boundaries are operational. They do not alter the semantic key or caller order. ### Phase 4: execute batches concurrently Up to `Workers` batches execute concurrently. Before one physical batch starts, every configured `Admitter` receives a `Work` value containing keys, attempt number, item count, and caller-defined units. The processor receives only typed inputs. It has no access to output positions or cache internals unless the application deliberately includes such data in `I`. ### Phase 5: commit successes immediately Each successful item is marked complete. When a cache exists, its encoded value is stored immediately under the semantic key. This matters when a later sibling or later retry fails: successful work is not lost merely because the whole call did not become globally successful. ### Phase 6: classify failures Every item failure is classified independently. Unknown errors fail closed as permanent. Only explicitly transient or rate-limited errors retry by default. If retry is permitted and attempts remain, the item enters the next wave. Successful siblings do not. ### Phase 7: wait once for the next wave The next delay is at least the policy backoff and any provider-supplied `RetryAfter`. The run waits before executing the pending set again. ### Phase 8: expand outcomes Each unique item outcome is copied to every original call position sharing its key. Caller order is restored exactly. ## 10.3 Stable order under concurrency Let input calls be $c_0,\ldots,c_{n-1}$. `Run` promises output outcomes $o_0,\ldots,o_{n-1}$ such that: $$ o_i.\operatorname{Key} = c_i.\operatorname{Key}. $$ This is independent of physical completion order. The package does not promise deterministic observer event order under multiple workers; that order is part of the operational trace. ## 10.4 Duplicate transparency For a set of call positions $D_k$ sharing key $k$, the executor performs one unique request and expands its outcome: $$ \forall i,j\in D_k,\quad (o_i.\operatorname{Value},o_i.\operatorname{Err}) = (o_j.\operatorname{Value},o_j.\operatorname{Err}). $$ This property is stronger than merely suppressing duplicate cache writes. It prevents duplicate in-flight work inside one run. ## 10.5 Partial retry Suppose a batch contains item keys $\{a,b,c\}$, and only $b$ fails transiently. The next wave contains $\{b\}$ rather than the entire batch. In set notation: $$ P_{r+1} = \{k\in P_r \mid \operatorname{failed}(k) \land \operatorname{retryable}(k) \land \operatorname{attempts}(k)