Skip to content
Documentation
Reference
Go SDK
journal
API

journal

import "github.com/danmestas/dagnats/journal"

journal/inputs.go Placement records (ADR-027). An input moves from the submissions bucket into the conversation by one input.placed record; the record is the truth about placement, so a journal knows every request_id placed in any iteration and a successor never places one twice.

Package journal is the agent journal of ADR-026: one append-only JetStream subject per step, journal.{run}.{step}, where an agent step’s SDK records what it did so a successor can take over without repeating a turn. Every append is fenced on the journal tail its writer last saw, so a worker that lost the tail (stalled past its AckWait, superseded by a redelivery) cannot write, and is told so.

A Journal is one claim on one dispatch of a step. Open reads the tail from the stream leader, stops without writing when the step has moved to a later iteration, rebuilds what this iteration needs from the records so far, and appends a claim record. Append and Commit then carry Nats-Expected-Last-Subject-Sequence and a stable Nats-Msg-Id. A rejected append is not fencing by itself: the writer re-reads the tail and only a record from another claim of the same or a later iteration fences it.

Index

Constants

Record types. Handlers write TypeEntry; the SDK writes the rest. The model and tool types are reserved for the helpers of #805.

const (
    TypeClaim         = "claim"
    TypePause         = "pause"
    TypeCommit        = "commit"
    TypeEntry         = "entry"
    TypeModelResponse = "model.response"
    TypeToolIntent    = "tool.intent"
    TypeToolResult    = "tool.result"
    TypeInputPlaced   = "input.placed"
)

Commit actions: what the iteration’s lifecycle event will say.

const (
    ActionContinue = "continue"
    ActionComplete = "complete"
    ActionFail     = "fail"
)

Record headers. The identity lives in headers so a tail read can classify a record without decoding its body.

const (
    HeaderType      = "Dagnats-Journal-Type"
    HeaderIteration = "Dagnats-Journal-Iteration"
    HeaderAttempt   = "Dagnats-Journal-Attempt"
    HeaderIndex     = "Dagnats-Journal-Index"
    HeaderDelivered = "Dagnats-Journal-Delivered"
    HeaderStreamSeq = "Dagnats-Journal-Stream-Seq"
    HeaderNonce     = "Dagnats-Dispatch-Nonce"
)

ErrInterrupted is the error text of a tool result recorded for a call a previous holder of the step started but never finished.

const ErrInterrupted = "interrupted"

Variables

Sentinel errors a handler is expected to check.

var (
    // ErrStale: the journal tail belongs to a later iteration, so this
    // task is a late redelivery. Nothing was written; the task is done.
    ErrStale = errors.New("journal: task is stale, a later iteration holds the step")
    // ErrFenced: a record from another claim follows this writer's.
    // Nothing more may be written or published for this task.
    ErrFenced = errors.New("journal: fenced by another claim")
    // ErrCommitted: this attempt already committed the iteration.
    // Open found the commit; the caller republishes its event.
    ErrCommitted = errors.New("journal: iteration already committed by this attempt")
    // ErrRestartLimit: the crash-loop guard failed the step (ADR-028).
    // The worker has published step.failed; the handler must return.
    ErrRestartLimit = errors.New("journal: restart limit reached, the step has been failed")
    // ErrNotProvisioned: the stream is missing.
    ErrNotProvisioned = errors.New(
        "journal: AGENT_JOURNAL stream not provisioned" +
            " — ensure natsutil.SetupAll() has been called" +
            " or start the server with 'dagnats serve'",
    )
)

ErrReplayMismatch: the call at this position was recorded with another name or input. The handler is not replaying the sequence of calls it recorded; nothing is reused or written.

var ErrReplayMismatch = errors.New("journal: tool call does not match the recorded call")

func MsgID 

func MsgID(task Task, index int) string

MsgID is the stable dedup key of a record: {run}.{step}.{attempt}.{iteration}.{stream_seq}.{delivered}.{index}. The delivery count keeps a redelivery’s records distinct from the stalled delivery’s inside the dedup window; the stream sequence does the same for a resend, which restarts the count. Without them the successor’s claim would be answered with the old claim’s sequence.

type Claim

Claim is the body of a claim record.

type Claim struct {
    Iteration     int    `json:"iteration"`
    Attempt       int    `json:"attempt"`
    DispatchNonce string `json:"dispatch_nonce,omitempty"`
    Delivered     int    `json:"delivered"`
    StreamSeq     uint64 `json:"stream_seq,omitempty"`
}

type Commit

Commit is the body of a commit record: the iteration’s outcome and the state the next iteration starts from.

type Commit struct {
    Action       string          `json:"action"`
    Output       json.RawMessage `json:"output,omitempty"`
    State        []byte          `json:"state,omitempty"`
    Error        string          `json:"error,omitempty"`
    FailureType  string          `json:"failure_type,omitempty"`
    RetryAfterMs int64           `json:"retry_after_ms,omitempty"`
    // Wait is the durable wait a continue asked for (ADR-027), so a
    // republished step.continue carries the same one.
    Wait *protocol.ContinueWait `json:"wait,omitempty"`
}

type InputPlaced

InputPlaced is the body of an input.placed record: the submission as it joined the conversation. Content is copied in so the conversation can be rebuilt from the journal alone, after the bucket’s TTL.

type InputPlaced struct {
    RequestID string          `json:"request_id"`
    Mode      string          `json:"mode"`
    Content   json.RawMessage `json:"content,omitempty"`
    // Iteration is set from the record on read; ignored on write.
    Iteration int `json:"-"`
}

type Journal

Journal is one claim on one dispatch of a step. A handle belongs to one goroutine: replay matches tool and model calls by position, so calls issued from several goroutines would replay in another order. A handler running a model’s parallel tool calls issues them through the journal one at a time. Concurrent use panics; a tool function may call back into the journal from the goroutine that owns it.

type Journal struct {
    // contains filtered or unexported fields
}

func Open 

func Open(ctx context.Context, js jetstream.JetStream, pub *natsutil.TracingPublisher, task Task) (*Journal, error)

Open claims task’s dispatch. It returns ErrStale when the tail belongs to a later iteration (nothing written), ErrFenced when a rival claim for the same iteration won, and a journal whose Committed reports this attempt’s own commit when the previous holder crashed between committing and publishing (no claim is appended then).

func (*Journal) Append 

func (j *Journal) Append(ctx context.Context, recordType string, data []byte) (uint64, error)

Append writes a record of recordType with data as its body.

func (*Journal) Commit 

func (j *Journal) Commit(ctx context.Context, c Commit) (uint64, error)

Commit ends the iteration with one record: the action its lifecycle event will carry and the state the next iteration starts from. One message, not an atomic batch (ADR-026). After it nothing more may be appended under this claim.

func (*Journal) Committed 

func (j *Journal) Committed() (Commit, bool)

Committed returns this attempt’s own commit: the one Open found, or the one Commit stored.

func (*Journal) CommittedSeq 

func (j *Journal) CommittedSeq() uint64

CommittedSeq returns the journal sequence of the commit Committed reports, 0 when there is none.

func (*Journal) Fenced 

func (j *Journal) Fenced() bool

Fenced reports whether a record from another claim fenced this one.

func (*Journal) InputPlaced 

func (j *Journal) InputPlaced(requestID string) bool

InputPlaced reports whether requestID has an input.placed record in this journal, in any iteration, by any claim.

func (*Journal) Model 

func (j *Journal) Model(ctx context.Context, fn ModelFunc) ([]byte, error)

Model requests one model response durably: the response a previous holder of this attempt recorded for the call at this position is returned without calling fn; otherwise fn runs and its response is appended as model.response. A request that fails records nothing and leaves the position free, and a new attempt of the iteration requests afresh. Streaming tokens are not the journal’s business: PutStream is a non-durable preview, and a crash mid-stream re-sends the request on the next delivery. Model must be called from the goroutine that owns the handle.

func (*Journal) Pause 

func (j *Journal) Pause(ctx context.Context, name string, duration time.Duration) (uint64, error)

Pause writes the pause record Pause publishes before its delayed redelivery; the next claim on this iteration sees it in Resume.

func (*Journal) PlaceInput 

func (j *Journal) PlaceInput(ctx context.Context, p InputPlaced) (uint64, error)

PlaceInput appends an input.placed record. The caller checks InputPlaced first; placing a request twice is a programmer error.

func (*Journal) PlacedInputs 

func (j *Journal) PlacedInputs() []InputPlaced

PlacedInputs returns every placement in journal order, with the iteration each joined.

func (*Journal) Records 

func (j *Journal) Records() []Record

Records returns this iteration’s records written before this claim, in journal order: what a recovering attempt may reuse.

func (*Journal) Restarts 

func (j *Journal) Restarts() int

Restarts returns how many claims on this iteration sat at the tail before this claim with nothing between them: deliveries that claimed and then made no progress. A pause, an entry, a tool or model record, or a commit resets the count; a claim after a commit (a retry) starts from 0. Feeds the worker’s restart limit (ADR-028).

func (*Journal) Resume 

func (j *Journal) Resume() (string, bool)

Resume returns the pause name when the newest record of this iteration before the claim is a pause: this delivery resumes it.

func (*Journal) Seq 

func (j *Journal) Seq() uint64

Seq returns the last journal sequence this writer saw.

func (*Journal) State 

func (j *Journal) State() []byte

State returns the state of the last commit before this iteration, nil when there is none.

func (*Journal) Task 

func (j *Journal) Task() Task

Task returns the dispatch this journal claims.

func (*Journal) Tool 

func (j *Journal) Tool(ctx context.Context, call ToolCall, replay Replay, fn ToolFunc) (ToolResult, error)

Tool runs one tool call durably: it appends the call’s intent with its replay declaration, runs fn, and appends the result. Calls are matched by position within the iteration, and a recorded call at the position must be the same call (name and input), else ErrReplayMismatch. On a claim that recovers an attempt a previous holder started: a recorded result is returned without running fn; an intent without a result is rerun only when the recorded and the current declaration both say Safe, and otherwise answered with an interrupted result carrying the partial output and the call’s key, which the model then sees. On a new attempt of the iteration (after a commit, or a higher attempt from iteration 1) only unsafe intents are honoured, matched by content rather than position: the first prior unsafe intent with the same name and input not yet honoured in this attempt answers the call under its own key, with its recorded result or as interrupted; safe calls run afresh.

The guarantee is that the SDK starts an unsafe call at most once per position. A holder stalled between its intent and the tool’s effect may still complete that effect after a successor recorded the call as interrupted; the key is what an external system deduplicates on. The returned error is the journal’s (ErrFenced, ErrCommitted, transport), never the tool’s own, which is in the result. Tool must be called from the goroutine that owns the handle; a model’s parallel tool calls are issued one at a time.

type ModelFunc

ModelFunc requests one model response. An error means no response: nothing is recorded and the next holder requests again.

type ModelFunc func(ctx context.Context) ([]byte, error)

type ModelResponse

ModelResponse is the body of a model.response record.

type ModelResponse struct {
    Call   int             `json:"call"`
    Output json.RawMessage `json:"output,omitempty"`
}

type Pause

Pause is the body of a pause record.

type Pause struct {
    Name       string `json:"name"`
    DurationMs int64  `json:"duration_ms"`
}

type Record

Record is one journal message.

type Record struct {
    Seq           uint64          `json:"seq"`
    Type          string          `json:"type"`
    Iteration     int             `json:"iteration"`
    Attempt       int             `json:"attempt"`
    DispatchNonce string          `json:"dispatch_nonce,omitempty"`
    Delivered     int             `json:"delivered"`
    StreamSeq     uint64          `json:"stream_seq,omitempty"`
    Index         int             `json:"index"`
    Time          time.Time       `json:"time"`
    Data          json.RawMessage `json:"data,omitempty"`
}

func DecodeMsg 

func DecodeMsg(msg jetstream.Msg) (Record, error)

DecodeMsg reads one journal message, as delivered by a consumer on the journal stream, into a Record. The record’s Seq is the message’s stream sequence, the cursor clients resume from.

func Read 

func Read(ctx context.Context, js jetstream.JetStream, runID, stepID string) ([]Record, error)

Read returns every record of the step’s journal, in order. An unprovisioned stream is ErrNotProvisioned; an empty journal is an empty slice.

type Replay

Replay is a tool’s declaration of what a rerun costs: Safe when running the call twice is harmless (a read, an idempotent write keyed on ToolCall.Key), Unsafe when it is not (send a message, open a pull request, charge a card).

type Replay string

const (
    Safe   Replay = "safe"
    Unsafe Replay = "unsafe"
)

type Report

Report is the result of Check for one run.

type Report struct {
    RunID string       `json:"run_id"`
    Steps []StepReport `json:"steps"`
    // Skipped names the invariants that could not be checked and why.
    Skipped []string `json:"skipped,omitempty"`
}

func Check 

func Check(ctx context.Context, nc *nats.Conn, runID string) (Report, error)

Check audits a run’s journals against its history and snapshot: each iteration has exactly one commit from the attempt the engine accepted (1); no record follows the claim that fenced its writer (2); every tool.intent has one tool.result (4); every placed input appears once (5); the engine’s iteration count equals the accepted continue commits (6). Invariant 3 needs a side-effect counter outside the journal and is reported as skipped, as is 5 while the submissions bucket does not exist.

func (Report) OK 

func (r Report) OK() bool

OK reports whether no step has a violation.

type ScriptedModel

ScriptedModel is a fake model for tests: it answers Model calls with its scripted responses in order and counts the requests it served, so a test can tell a reused response from a re-requested one. Safe for concurrent use.

type ScriptedModel struct {
    // contains filtered or unexported fields
}

func NewScriptedModel 

func NewScriptedModel(responses ...[]byte) *ScriptedModel

NewScriptedModel scripts the responses, in order.

func (*ScriptedModel) Requests 

func (m *ScriptedModel) Requests() int

Requests is how many responses the model has served.

func (*ScriptedModel) Respond 

func (m *ScriptedModel) Respond(ctx context.Context) ([]byte, error)

Respond is the ModelFunc: the next scripted response, or an error once the script is exhausted.

type StepReport

StepReport is one step’s audit.

type StepReport struct {
    StepID     string   `json:"step_id"`
    Records    int      `json:"records"`
    Commits    int      `json:"commits"`
    Violations []string `json:"violations,omitempty"`
}

type Task

Task identifies one dispatch of a step: the identity every record carries and every lifecycle event names (ADR-025).

type Task struct {
    RunID         string
    StepID        string
    Iteration     int
    Attempt       int
    DispatchNonce string
    // Delivered is the delivery count of the task message. It is part
    // of the claim identity: the engine stamps the attempt on every
    // dispatch, so a stalled worker and the AckWait redelivery that
    // replaces it share run, step, iteration, attempt and nonce, and
    // only the delivery count tells their claims apart. It also orders
    // the deliveries of one message (a newer one fences an older one
    // opening after it) and feeds the crash-loop guard (#807).
    Delivered int
    // StreamSeq is the task message's sequence in the task stream.
    // Redeliveries share it; a dispatch resend is a fresh message with
    // a new one and delivery count 1 again, so only it tells a resend's
    // claim from the original's while that is still in flight. Zero
    // when unknown.
    StreamSeq uint64
}

type ToolCall

ToolCall is one tool invocation. The handler sets Name and Input; Tool fills Call (the call’s position in the iteration, counting from 0) and Key, the idempotency key {run}.{step}.{iteration}.{call}, which is the same on every delivery and attempt of the iteration so the tool can hand it to an external system that deduplicates.

type ToolCall struct {
    Name  string          `json:"name"`
    Input json.RawMessage `json:"input,omitempty"`
    Call  int             `json:"call"`
    Key   string          `json:"key"`
}

type ToolFunc

ToolFunc runs the tool. The output it returns must be a JSON value, because it is recorded as is: encode plain text with json.Marshal. The error it returns is the tool’s own failure, recorded in the result for the model to see; it is not a reason to stop the step.

type ToolFunc func(ctx context.Context, call ToolCall) ([]byte, error)

type ToolIntent

ToolIntent is the body of a tool.intent record: the call and its replay declaration, written before the tool runs.

type ToolIntent struct {
    ToolCall
    Replay Replay `json:"replay"`
}

type ToolResult

ToolResult is the body of a tool.result record. Key is the call’s idempotency key. Interrupted marks a call a previous holder started and never finished, which was not rerun; Partial then holds, as a JSON array, the entry records that holder appended while the tool ran.

type ToolResult struct {
    Call        int             `json:"call"`
    Key         string          `json:"key"`
    Output      json.RawMessage `json:"output,omitempty"`
    Error       string          `json:"error,omitempty"`
    Interrupted bool            `json:"interrupted,omitempty"`
    Partial     json.RawMessage `json:"partial,omitempty"`
}

Generated by gomarkdoc