internal/agent

import "github.com/neochaotic/leoflow/internal/agent"

const DefaultHeartbeatInterval = 15 * time.Second

func AttemptTokenTTL(interval time.Duration) time.Duration

func Backoff(attempt int) (delay time.Duration, ok bool)

func BuildCommand(operator, entrypoint, operatorClass string) ([]string, error)

func ClassifyBootstrapFailure(stage BootstrapStage, err error) string

func ExchangeToken(ctx context.Context, client agentv1.AgentServiceClient, tokens *TokenSource) error

func NewReturnValuePath() (path string, cleanup func() error, err error)

func ReadReturnValue(path string) (value []byte, ok bool, err error)

func ReadTokenFile(path string) (string, error)

func ReportBootstrapFailure(path string, stage BootstrapStage, err error)

func XComEnvVar(name string, value []byte) string

type BootstrapStage int

const (
    // StageToken is reading the projected ServiceAccount token from the pod.
    StageToken BootstrapStage = iota
    // StageDial is establishing the gRPC channel to the control plane.
    StageDial
    // StageExchange is trading the bootstrap token for a task-scoped credential.
    StageExchange
)

type CommandRunner interface {
    Run(ctx context.Context, argv, env []string, stdout, stderr io.Writer) (exitCode int, err error)
}

func NewExecRunner() CommandRunner

type LogSink interface {
    Send(line *agentv1.LogLine) error
    Close() error
}

func OpenLogSink(ctx context.Context, client agentv1.AgentServiceClient) (LogSink, error)

type NoopLogSink struct{}

func (NoopLogSink) Close() error

func (NoopLogSink) Send(*agentv1.LogLine) error

type Runner struct {
    Client     agentv1.AgentServiceClient
    Cmd        CommandRunner
    Sink       LogSink
    Hostname   string
    Version    string
    Env        []string // base process environment (typically os.Environ())
    ReturnPath string   // file the task writes its return value to; empty disables push
    LinksPath  string   // file the runtime writes operator_extra_links to; empty disables (#375)
    PushesPath string   // file the runtime writes custom-keyed XCom pushes to; empty disables (multi-key XCom)
    // ReschedulePath is the file a reschedule-mode sensor writes its next-poke time
    // to before exiting with rescheduleExitCode; empty disables reschedule (#380).
    ReschedulePath string
    // TmpDir, when set, is exported to the task as TMPDIR so the child's temp files
    // land in a per-attempt directory the caller wipes between attempts, instead of
    // the pod's real /tmp. The warm worker points this at a subdir of its scratch
    // (reset before every attempt) so no attempt observes another attempt's temp
    // files — a token cache, a dbt profile, ~/.aws-style credentials (#728). Empty
    // leaves TMPDIR untouched: a single-shot pod is already destroyed per task, so
    // its /tmp needs no in-process reset.
    TmpDir string
    // TerminationLogPath is where the agent writes its durable outcome record just
    // before delivering the report, so a pod killed mid-report still leaves the
    // task's true result behind for the reconciler to recover (ADR 0052). Empty
    // disables it — Lite (subprocess, in-process report) needs no such record.
    TerminationLogPath string
    // HeartbeatInterval is how often to ping the control plane while the task
    // runs; zero disables heartbeats.
    HeartbeatInterval time.Duration
    // Token, when set, is the swappable bearer backing the gRPC per-RPC credential.
    // On a heartbeat carrying a renewed_token the loop atomically swaps it here so
    // every subsequent RPC uses the new credential (ADR 0055 Fix #4). Nil disables
    // bearer swapping (a heartbeat's renewed_token is then ignored). Typed as the
    // narrow tokenSetter seam (satisfied by *TokenSource) so the heartbeat's Set
    // can be observed in tests; production always wires a *TokenSource.
    Token tokenSetter
    // BeforeReport, if set, is invoked with the terminal state AFTER the durable
    // outcome record is written and BEFORE the report is delivered. It is a
    // fault-injection seam for the durable-outcome E2E (ADR 0052) — the agent
    // binary wires it, from an env var, to exit the process, simulating a pod
    // killed mid-report with the record already on disk. Nil in production.
    BeforeReport func(agentv1.TaskState)
    // contains filtered or unexported fields
}

func (r *Runner) Run(ctx context.Context) error

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

func Dial(addr, token string, allowInsecure bool, caFile string) (agentv1.AgentServiceClient, *grpc.ClientConn, *TokenSource, error)

func NewTokenSource(token string) *TokenSource

func (s *TokenSource) Set(token string)

func (s *TokenSource) Token() string

type WarmRunner struct {
    StreamClient  agentv1.AgentServiceClient
    WorkClient    agentv1.AgentServiceClient
    AttemptTokens *TokenSource

    // StreamTokens is the bootstrap stream's own TokenSource (the source
    // StreamClient's per-RPC credential reads). Under the exchange transport it is
    // seeded with the projected ServiceAccount token; the exchange step below swaps
    // the WORKER-scoped JWT into it before Register, so Register + AwaitAssignment
    // carry the worker credential. Distinct from AttemptTokens (the WorkClient's
    // per-attempt source). Unused under the env-var transport.
    StreamTokens *TokenSource

    // ExchangeBootstrap runs the projected-token → worker-JWT exchange (ADR 0058 D2)
    // on StreamClient before Register, on the INITIAL connect and on every reconnect,
    // swapping the minted worker JWT into StreamTokens. It is set under the exchange
    // transport; false under the env-var default (Register uses the seed token as-is).
    ExchangeBootstrap bool

    // NewSink opens a fresh per-attempt log sink on the WorkClient, so each
    // attempt's logs are shipped under its own attempt_token. Nil (or a returned
    // error) falls back to NoopLogSink — logs are best-effort, never fatal.
    NewSink func(ctx context.Context) (LogSink, error)

    Cmd      CommandRunner
    Hostname string
    Version  string
    Env      []string // base process environment (typically os.Environ())

    // PodName is the worker's OWN Kubernetes pod name, read from LEOFLOW_POD_NAME
    // (injected via the downward API) in main.go. It is sent up in WorkerRegister
    // so the control plane can bind a started attempt to it as the durable
    // warm_worker_id (ADR 0058 N1d-a1). Empty outside Kubernetes (e.g. tests); the
    // binding then simply degrades to per-pod liveness for this worker.
    PodName string

    // ScratchDir is the agent-owned per-attempt scratch root. It is wiped and
    // recreated before every attempt (D4 isolation) and holds the return-value,
    // extra-links, xcom-pushes, and reschedule files the runtime writes.
    ScratchDir string

    // TerminationLogPath and HeartbeatInterval mirror the single-shot Runner's
    // fields and are threaded into every per-attempt Runner.
    TerminationLogPath string
    HeartbeatInterval  time.Duration

    // Self-lifecycle bounds (ADR 0058 D9/D10/D6/H3), populated from the warm-pod env
    // in main.go. A warm worker that exits is replaced by the reconciler
    // (RestartPolicy:Never + busy-aware create), so bounding its own life is how a
    // pool stays fresh and scales down. Each bound is disabled when zero/unset — a
    // defensive default; the operator config values are non-zero.
    //
    //   - MaxAttempts: drain after this many completed attempts (D9/D10).
    //   - MaxLifetime: drain once the worker is this old (D9/D10).
    //   - IdleTTL: idle-recycle after this long awaiting the next assignment (D6).
    //   - AttemptWatchdog: hard per-attempt ceiling, INDEPENDENT of the task's
    //     execution_timeout, so a task that declares no timeout and then wedges is
    //     still killed and the slot freed (H3).
    MaxAttempts     int
    MaxLifetime     time.Duration
    IdleTTL         time.Duration
    AttemptWatchdog time.Duration

    // Reconnect-toward-the-leader bounds (warm-pool Hole B). On a not-leader
    // rejection (FailedPrecondition — the leader-gate refusal, or warm-pools not
    // serving on this endpoint) the worker RECONNECTS with jittered exponential
    // backoff instead of exiting: a single pod finds the leader across the Service's
    // rotation without the pod-create-die-recreate churn a fresh pod would cause. The
    // reconnect is bounded so a genuinely misconfigured deployment eventually exits
    // (non-zero) and lets the reconciler replace it, rather than spinning forever.
    // Zero values fall back to production defaults; tests inject tiny ones.
    //
    //   - ReconnectBackoff: base backoff, doubled each consecutive reconnect.
    //   - ReconnectMaxBackoff: ceiling on a single backoff sleep.
    //   - MaxReconnects: consecutive not-leader reconnects before giving up.
    ReconnectBackoff    time.Duration
    ReconnectMaxBackoff time.Duration
    MaxReconnects       int

    // Redial re-establishes the StreamClient toward the control plane for the next
    // reconnect. A FRESH dial is what lets the worker reach the leader: a single gRPC
    // connection sticks to one Service backend, so retrying the same client would
    // re-hit the same follower forever — only a new connection re-rotates. It returns
    // the new stream client, its TokenSource (so an exchange-transport worker
    // re-exchanges into the fresh connection's bearer), and a closer for the
    // connection it opened (closed on the next reconnect / exit). Nil disables
    // re-dial — the reconnect then retries the existing StreamClient (tests inject a
    // fake that eventually serves); production wires a real re-dial via agent.Dial.
    Redial func() (agentv1.AgentServiceClient, *TokenSource, io.Closer, error)
    // contains filtered or unexported fields
}

func (w *WarmRunner) Run(ctx context.Context, dagVersionID string) error