taskrun

package
v0.15.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 3, 2026 License: AGPL-3.0 Imports: 33 Imported by: 0

Documentation

Overview

Package taskrun orchestrates coding-agent state and ordered conversation and Git history.

It owns branch creation, instance lifecycle, agent execution, resource history, and git integration.

Index

Constants

This section is empty.

Variables

View Source 
var ErrInitialPromptNotAccepted = errors.New("task startup did not accept its initial prompt")

ErrInitialPromptNotAccepted reports that startup ended before the task's first user input became part of the durable conversation.

View Source 
var ErrNoActiveSession = errors.New("no active session")

ErrNoActiveSession reports that input cannot be delivered because no live session is attached to the task.

Functions

func BuildHandoffPrompt 

func BuildHandoffPrompt(source *Task, maxBytes int) string

BuildHandoffPrompt renders task context for a fresh harness session.

Only direct user and assistant text is included from the conversation. Thinking, tool activity, raw output, and other private operational messages are omitted. A non-positive maxBytes uses a conservative default.

func MakeMetadata 

func MakeMetadata(t *Task) runtime.Metadata

MakeMetadata builds runtime metadata for a task instance.

MetadataLegacyTaskID is kept alongside MetadataTaskID for compatibility with existing instances and event filters. Once old instances have cycled out, the legacy metadata key can be removed.

Types

type AgentRuntime 

type AgentRuntime struct {
	// Immutable.
	Backends         agent.Backends
	LogStore         *taskhistory.Store
	LogPath          *taskslog.Path
	Runtimes         *runtime.Router
	Log              *slog.Logger
	NotifyTaskChange func()

	Checkout             *repo.Checkout // nil for no-repository tasks
	RuntimeMetadata      runtime.Metadata
	RuntimeStartTimeout  time.Duration // Timeout for instance start (image pull). Must be non-zero.
	EnsureSessionTimeout time.Duration // Timeout for session verification in EnsureSession. Defaults to 10s if <= 0.
	MCPRegistry          mcp.Registry  // Task-scoped CAIC MCP registry; nil when disabled.
}

AgentRuntime owns task runtime setup, agent sessions, log persistence, message dispatch, cleanup, restart, reconnect, revive, and fork operations.

func (*AgentRuntime) Cancel 

func (r *AgentRuntime) Cancel(t *Task)

Cancel stops future execution admission and interrupts existing local producers without waiting or closing recording. Cancellation is repeatable; Shutdown is the single joining and physical-close operation.

func (*AgentRuntime) Cleanup 

func (r *AgentRuntime) Cleanup(ctx context.Context, t *Task, reason task.State) task.Result

Cleanup is the single shutdown path for a task (Flow 1 in the relay shutdown protocol — see package agent). It sends the null-byte sentinel to trigger graceful agent exit, then purges the instance.

This is only called for intentional purge (user action or instance death), never during backend restart. On restart, the relay daemon stays alive and the server reconnects during task import.

Steps:

  1. Detach the session handle from the task.
  2. If a session exists: Stop (sends \x00, waits up to 20s), then Close.
  3. Set task state to reason (StatePurged or StateFailed).
  4. Purge the instance (stop + remove + cleanup git remotes/runtime config).
  5. If graceful wait timed out, drain session now (runtime connection severed).
  6. Close msgCh and log, write log trailer.
  7. Build and return Result.

func (*AgentRuntime) ClearContextSession 

func (r *AgentRuntime) ClearContextSession(ctx context.Context, t *Task) (*SessionHandle, error)

ClearContextSession closes the current agent session and starts a fresh one in the same instance without a prompt. The task transitions to StateWaiting so the user can send a new message when ready.

func (*AgentRuntime) CommitSettledResult 

func (r *AgentRuntime) CommitSettledResult(ctx context.Context, t *Task, res *task.Result) error

CommitSettledResult records a result only when no execution owns the journal.

func (*AgentRuntime) EnsureSession 

func (r *AgentRuntime) EnsureSession(ctx context.Context, tlog *slog.Logger, t *Task, h *SessionHandle) (*SessionHandle, error)

EnsureSession waits briefly for h to confirm it's alive. If the session exits within 10 seconds (agent had already finished), it detaches and starts a fresh idle relay only when the task has no saved session. Existing conversations fail instead of silently losing history. The supplied context must also own the session process; cancellation drains its local transport without sending a stop sentinel to the retained remote agent.

func (*AgentRuntime) FinishSession 

func (r *AgentRuntime) FinishSession(ctx context.Context, t *Task, h *SessionHandle, sessionErr error) (bool, error)

FinishSession claims the expected execution after its transport has joined. Operational failures affect runtime health without inventing agent messages.

func (*AgentRuntime) ForkTask 

func (r *AgentRuntime) ForkTask(ctx context.Context, source, fork *Task, forkOpts *runtime.ForkOptions, resolvedGitHubToken string) (*SessionHandle, error)

ForkTask snapshots the source task's instance and starts an idle agent session in the forked instance. The new task must already have its ID, Harness, Model, and other immutable fields set. The method fills in Runtime, Repos[*].Branch, and starts the session.

func (*AgentRuntime) Reconnect 

func (r *AgentRuntime) Reconnect(ctx context.Context, t *Task) (*SessionHandle, error)

Reconnect attaches to the existing relay using validated persistent history. A dead relay returns an error; startup recovery separately revives eligible retained instances rather than replacing an unverified live relay.

func (*AgentRuntime) RecordHost 

func (r *AgentRuntime) RecordHost(ctx context.Context, t *Task, m agent.Message) error

RecordHost records an intentional mutation through the active owner, or an exclusive short-lived owner when the execution has settled.

func (*AgentRuntime) RecordLifecycleIntent 

func (r *AgentRuntime) RecordLifecycleIntent(ctx context.Context, t *Task, persistedState, runtimeState task.State, allowedStates ...task.State) (task.State, bool, error)

RecordLifecycleIntent persists an accepted state intent before publishing its runtime transition. Rejected intents publish neither records nor transitions.

func (*AgentRuntime) RecordRepositorySummary 

func (r *AgentRuntime) RecordRepositorySummary(ctx context.Context, t *Task, read *repo.GitSnapshot) (bool, error)

RecordRepositorySummary admits a completed Git read in physical stream order.

func (*AgentRuntime) RecoverRetainedSession 

func (r *AgentRuntime) RecoverRetainedSession(ctx context.Context, t *Task, expectedState task.State) (*SessionHandle, error)

RecoverRetainedSession records recovery intent before stopping an unverified retained runtime, then resumes it through the same newly owned execution.

func (*AgentRuntime) RestartSession 

func (r *AgentRuntime) RestartSession(ctx context.Context, t *Task, prompt agent.Prompt) (*SessionHandle, error)

RestartSession closes the current agent session and starts a fresh one in the same instance with a new prompt. Returns the new SessionHandle so the caller can start a session watcher.

func (*AgentRuntime) ReviveTask 

func (r *AgentRuntime) ReviveTask(ctx context.Context, t *Task) (*SessionHandle, error)

ReviveTask restarts a stopped or crashed instance and resumes the agent session. The instance's filesystem is preserved from the previous run.

func (*AgentRuntime) Shutdown 

func (r *AgentRuntime) Shutdown(ctx context.Context, t *Task) error

Shutdown cancels and joins local producers without stopping a retained remote relay. It also joins startup before claiming and closing the record owner.

func (*AgentRuntime) Start 

func (r *AgentRuntime) Start(ctx context.Context, t *Task, resolvedGitHubToken string) (*SessionHandle, error)

Start performs branch/instance setup, starts the agent session, and sends the initial prompt. Returns the SessionHandle so the caller can start a session watcher.

Sequence:

  1. Adopt the selected unused local branch, or create a new branch from the selected remote/default branch.
  2. Start a runtime instance on that branch.
  3. Deploy the relay script and launch the agent (claude) via the relay daemon. The relay owns the agent's stdin/stdout and persists across transport disconnects.
  4. Send the initial prompt to the agent.

The session is left open for follow-up messages via SendInput.

func (*AgentRuntime) StartSession 

func (r *AgentRuntime) StartSession(ctx context.Context, t *Task, prompt agent.Prompt) (*SessionHandle, error)

StartSession starts a fresh relay+agent session on an existing instance. If prompt is non-empty, it is sent as the initial input and the task transitions to StateRunning. If prompt is empty, the agent starts idle and the task stays in its current state (typically StateWaiting).

func (*AgentRuntime) StopTask 

func (r *AgentRuntime) StopTask(ctx context.Context, t *Task) *task.Result

StopTask gracefully shuts down the agent session and stops the instance without removing it. The instance can be revived later. Unlike Cleanup, this preserves git remotes and runtime config. It returns the persisted result, or nil when a concurrent teardown supersedes the stop.

type RateLimit 

type RateLimit struct {
	Status          agent.RateLimitStatus
	ResetsAt        time.Time
	RateLimitType   string  // Harness-native window ID; QuotaWindow is the canonical provider window.
	Utilization     float64 // Fraction of the window used in [0, 1]; -1 when unknown.
	IsUsingOverage  bool
	OverageResetsAt time.Time
	QuotaProvider   agent.QuotaProvider // Canonical usage-provider ID matching ProviderQuota.Provider.
	QuotaLabel      string
	QuotaWindow     string
	ObservedAt      time.Time
}

RateLimit records the active quota block resolved for a task. It is retained with task state so summaries and provider monitors can react immediately without waiting for a separate provider-usage refresh.

type Run 

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

Run owns an execution's ordered records before and after transport startup. Its creator joins startup; the expected-identity claimant joins all producers and closes recording once. Task.recordingMu serializes those claims.

func (*Run) ApplyRecords 

func (r *Run) ApplyRecords(ctx context.Context, messages []agent.TimedMessage) error

ApplyRecords folds only committed messages and starts owned title work.

func (*Run) PrepareRecords 

func (r *Run) PrepareRecords(_ context.Context, messages []agent.TimedMessage) ([]agent.Message, error)

PrepareRecords completes derived Git work at the finished turn's barrier. No task mutation or publication occurs before its controls have appended.

func (*Run) RecordAppended 

func (r *Run) RecordAppended(record agent.PhysicalRecord)

RecordAppended advances the remote cursor only after physical acceptance.

type SessionHandle 

type SessionHandle struct {
	Session *agent.Session
	Run     *Run
}

SessionHandle associates a transport with its owning execution lifetime.

func (*SessionHandle) Done 

func (h *SessionHandle) Done() <-chan struct{}

Done closes after the local transport process exits.

func (*SessionHandle) Drain 

func (h *SessionHandle) Drain() error

Drain joins the transport's reader, input, and protocol workers.

func (*SessionHandle) GracefulStop 

func (h *SessionHandle) GracefulStop(ctx context.Context, timeout time.Duration) error

GracefulStop stops the retained remote agent and waits up to timeout.

type SessionStatus 

type SessionStatus string

SessionStatus describes why SendInput could not deliver a message.

Session lifecycle:

  • A session wraps an SSH process bridging the server to the in-instance relay daemon. It is set by Checkout.Start, Checkout.Reconnect, or Checkout.RestartSession.
  • The session is cleared by CloseSession (during restart), Kill (during purge), or lazily by SendInput when it detects the SSH process already exited (Done channel closed).
  • "none" means no session was ever attached for this task — either the task hasn't started, or the relay died and reconnect failed.
  • "exited" means a session existed but the underlying SSH process exited (relay or agent crashed, SSH dropped) before the user sent input.
const (
	// SessionNone indicates no session was set on the task.
	SessionNone SessionStatus = "none"
	// SessionExited indicates the session's SSH process had already exited.
	SessionExited SessionStatus = "exited"
)

type Snapshot 

type Snapshot struct {
	agent.SessionMetadata

	State              task.State
	StateUpdatedAt     time.Time
	TurnStartedAt      time.Time // non-zero only while state is Running
	Repos              []repo.Mount
	RuntimeName        runtime.Name
	RuntimeInstanceID  runtime.ID
	Tailscale          bool
	TailscaleFQDN      string
	TailscaleAuthURL   string
	USB                bool
	Display            bool
	Sudo               bool
	SudoPassword       string
	VNCPort            int // VNC WebSocket port (0 = no VNC).
	GitHubToken        bool
	RelayOffset        int64
	Title              string
	RequestedModel     string
	RequestedEffort    string
	ContextWindowLimit int // Non-zero when reported by the agent at runtime.
	InPlanMode         bool
	PlanFile           string
	PlanContent        string
	PlanExitID         string
	CostUSD            float64
	NumTurns           int
	Duration           time.Duration
	Usage              agent.Usage
	LastUsage          agent.Usage
	LastAPIUsage       agent.Usage
	CacheExpiresAt     time.Time
	DiffStat           v3.DiffStat
	RepoStates         []v3.RepoState // Compact per-repo git state from the latest diff probe.
	DiskUsed           int64
	DiskKnown          bool
	ForgeOwner         string
	ForgeRepo          string
	ForgePR            int
	ForgePRState       forge.PRState // "open", "closed", "merged"; empty when no PR.
	ForgeIssue         int
	CIStatus           forge.CIStatus
	CIChecks           []forge.Check
	RateLimit          RateLimit
}

Snapshot holds volatile task fields read under the mutex. Used by the server to build API responses without data races on fields that ordered ingestion and SeedTimelineParts modify concurrently.

type StartupError 

type StartupError struct {
	Harness harness.Name
	Phase   string
	Err     error
}

StartupError identifies a task startup failure without discarding its cause. It is persisted in the task result and exposed to clients as structured data.

func (*StartupError) Details 

func (e *StartupError) Details() v3.StartupFailure

Details returns the durable client-facing startup diagnostic.

func (*StartupError) Error 

func (e *StartupError) Error() string

Error implements error.

func (*StartupError) Unwrap 

func (e *StartupError) Unwrap() error

Unwrap returns the underlying startup diagnostic.

type StateTransition 

type StateTransition struct {
	Seq   uint64
	State task.State
	At    time.Time
}

StateTransition records one task state change in the bounded journal used to replay short-lived states to snapshot-diffing clients. Seq increases monotonically per task and is never reused.

type Task 

type Task struct {

	// Immutable fields — set at creation, never modified.
	ID                ksid.ID
	InitialPrompt     agent.Prompt         // Initial prompt text and optional images.
	Harness           harness.Name         // Agent harness ("claude", "codex", etc.).
	RequestedModel    string               // User-requested model; passed to agent CLI.
	RequestedEffort   string               // Thinking effort; passed to agent CLI. Empty = default.
	RuntimeName       runtime.Name         // Runtime backend used for this task.
	BaseImage         string               // Custom runtime base image; empty means use the default.
	ContainerPlatform string               // Container CPU architecture; empty means use the host default.
	MaxCPUs           int                  // Max CPU cores for the instance; 0 means use the default.
	CacheMounts       []runtime.CacheMount // Build cache mounts baked into the runtime image.
	Mounts            []runtime.Mount      // Host directories bind-mounted into the runtime instance.
	Tailscale         bool                 // Enable Tailscale networking in the instance.
	USB               bool                 // Enable USB passthrough in the instance.
	Display           bool                 // Enable Xvfb display in the instance.
	Sudo              bool                 // Enable root access (password-based sudo) in the instance.
	StartedAt         time.Time            // When the task was created.
	OwnerID           string               // Human authorization principal; distinct from the delegating task's identity.
	ForgeIssue        int                  // Originating issue number for bot comment callbacks; 0 = none.
	ForkedFromTaskID  ksid.ID              // Parent task ID when created by fork; zero otherwise.
	ParentTaskID      ksid.ID              // Delegating task ID for a child task; zero for root tasks and ordinary forks.
	CaicMCP           bool                 // Enables task-scoped CAIC MCP delegation.
	Pricer            usage.ModelPricer    // Quota-provider per-model pricing for cost reporting; set by the task manager before the task runs. Nil keeps harness-reported totals only.
	Rollup            task.RollupSink      // Cross-task usage rollup sink; NewTask defaults to DiscardRollup, the task manager replaces it with the wired store.
	Provider          genai.Provider

	// Mutable task metadata. These fields are populated at construction, setup, or
	// import. After a task is published in the Manager registry, access them
	// through Task methods so readers and async lifecycle goroutines synchronize.
	Repos            []repo.Mount // index 0 = primary; empty = no-repo
	TailscaleFQDN    string       // Tailscale FQDN assigned to the instance (empty if not available).
	TailscaleAuthURL string       // Tailscale browser auth URL when no pre-auth key was available.
	RelayOffset      int64        // Committed relay output bytes in the current generation, for reconnect.
	SudoPassword     string       // Random sudo password; empty if sudo is not enabled.
	VNCPort          int          // VNC WebSocket port inside the instance (0 = no VNC). Set during launch.
	GitHubToken      bool         // Inject GitHub token into the instance's environment.
	// contains filtered or unexported fields
}

Task represents a single unit of work.

func NewTask 

func NewTask(id ksid.ID, prompt agent.Prompt, h harness.Name, model, effort, baseImage, containerPlatform, title string) (*Task, error)

NewTask creates a task with a valid ID, a pending state, and prompt.

func (*Task) AttachSession 

func (t *Task) AttachSession(h *SessionHandle)

AttachSession stores the SessionHandle under the mutex.

func (*Task) BackwardMessages 

func (t *Task) BackwardMessages() iter.Seq[agent.Message]

BackwardMessages returns the task's messages newest first as a snapshot taken when BackwardMessages is called. The sequence must be ranged exactly once so its snapshot lease is released; a second range is a no-op. Messages appended afterward are excluded, and finalized deltas are not compacted during the range.

func (*Task) CloseAndDetachSession 

func (t *Task) CloseAndDetachSession(ctx context.Context) (*SessionHandle, error)

CloseAndDetachSession gracefully shuts down the current agent session and returns the detached handle. Returns nil if no session was attached. Used by RestartSession which needs the graceful drain before starting a new session.

func (*Task) DetachSession 

func (t *Task) DetachSession() *SessionHandle

DetachSession atomically removes and returns the current SessionHandle, or nil if no session is attached. The caller must not hold t.mu.

func (*Task) DiffCreated 

func (t *Task) DiffCreated() bool

DiffCreated reports whether any non-empty diff was observed for the task.

func (*Task) DiscardRollup 

func (t *Task) DiscardRollup()

DiscardRollup stops forwarding this task's events and tells its previous sink to forget unflushed usage and replay state. It is called once a purge begins, before cleanup can receive late agent messages.

func (*Task) DiskUsage 

func (t *Task) DiskUsage() (int64, bool)

DiskUsage returns the latest writable-layer measurement, when one is known.

func (*Task) ExtraRuntimeRepos 

func (t *Task) ExtraRuntimeRepos() []runtime.Repo

ExtraRuntimeRepos returns all repos after the primary.

func (*Task) ForwardMessages 

func (t *Task) ForwardMessages() iter.Seq[agent.Message]

ForwardMessages returns the task's messages oldest first as a snapshot taken when ForwardMessages is called. The sequence must be ranged exactly once so its snapshot lease is released; a second range is a no-op. Messages appended afterward are excluded, and finalized deltas are not compacted during the range.

func (*Task) GenerateTitle 

func (t *Task) GenerateTitle(ctx context.Context, log *slog.Logger)

GenerateTitle asks the LLM for a short title from the last ten root turns, making up to two shortening requests when the title exceeds the character limit and logging through the task-scoped log. No-op when the provider is unconfigured.

func (*Task) GetPR 

func (t *Task) GetPR() int

GetPR returns the forge PR number (0 if no PR has been created).

func (*Task) GetPlanFile 

func (t *Task) GetPlanFile() string

GetPlanFile returns the plan file path under the mutex.

func (*Task) GetSessionID 

func (t *Task) GetSessionID() string

GetSessionID returns the agent session ID under the mutex.

func (*Task) GetState 

func (t *Task) GetState() task.State

GetState returns the current state under the mutex.

func (*Task) GitHubTokenEnabled 

func (t *Task) GitHubTokenEnabled() bool

GitHubTokenEnabled reports whether this task injects a GitHub token.

func (*Task) GitTarget 

func (t *Task) GitTarget() repo.GitTarget

GitTarget captures the instance and repository mapping in one task read.

func (*Task) GracefulStopSession 

func (t *Task) GracefulStopSession(ctx context.Context, timeout time.Duration) (*SessionHandle, error)

GracefulStopSession detaches the currently attached session (if any) and sends the relay's shutdown sentinel, waiting up to timeout for the agent to exit gracefully. Returns the detached handle (nil if no session was attached) and any error from the graceful stop, e.g. a timeout. Used by callers (Cleanup, StopTask) that need the handle afterward to reuse its Run for terminal recording after transport producers have joined.

func (*Task) HasAcceptedInputEvidence 

func (t *Task) HasAcceptedInputEvidence() bool

HasAcceptedInputEvidence reports whether durable conversation history proves that a user prompt reached the harness. Marked input attempts require a completed delivery outcome or real root output. Unmarked historical inputs retain their established meaning.

func (*Task) HasSession 

func (t *Task) HasSession() bool

HasSession reports whether a session handle is attached.

func (*Task) LastAgentResult 

func (t *Task) LastAgentResult() string

LastAgentResult returns the result text from the most recent ResultMessage, falling back to the visible text of that turn when the agent reported none, or "" if no result was received. Used by Cleanup/StopTask for the log trailer.

func (*Task) LastExitError 

func (t *Task) LastExitError() string

LastExitError returns the most recent non-zero relay exit diagnostic.

func (*Task) LiveDiffStat 

func (t *Task) LiveDiffStat() v3.DiffStat

LiveDiffStat returns the latest diff stat from the relay's periodic polling.

func (*Task) LiveStats 

func (t *Task) LiveStats() (costUSD float64, numTurns int, duration time.Duration, cumulativeUsage, lastTurnUsage agent.Usage)

LiveStats returns the latest cost, turn count, duration, cumulative token usage, and the most recent turn's usage (active context).

func (*Task) LogFilename 

func (t *Task) LogFilename() string

LogFilename returns the canonical filename for a new task-log segment.

func (*Task) LogHeader 

func (t *Task) LogHeader() *agent.MetaMessage

LogHeader builds the immutable metadata header for a new task-log segment.

func (*Task) MarkDiffCreated 

func (t *Task) MarkDiffCreated()

MarkDiffCreated records that a non-empty diff was observed, without a diff stat payload. Import uses it to restore the persisted DiffCreated flag from the log summary, so the signal survives a restart even when message replay is skipped or the relay's last diff was empty. The flag is sticky and never cleared here.

func (*Task) MarkSessionCrashed 

func (t *Task) MarkSessionCrashed() bool

MarkSessionCrashed records operational health when a recoverable session ends. The execution owner separately reports its error through the lifecycle result.

func (*Task) MarkSessionFailed 

func (t *Task) MarkSessionFailed() bool

MarkSessionFailed records operational health when startup transport fails. The execution owner separately reports its error through the lifecycle result.

func (*Task) MessageCount 

func (t *Task) MessageCount() int

MessageCount returns the number of messages retained in the task timeline.

func (*Task) Messages 

func (t *Task) Messages() []agent.Message

Messages returns a copy of all received agent messages.

func (*Task) PendingUserActions 

func (t *Task) PendingUserActions() []v3.PendingUserAction

PendingUserActions returns current user-facing actions that still need input.

func (*Task) Primary 

func (t *Task) Primary() *repo.Mount

Primary returns a copy of the primary repository mount (Repos[0]), or nil for no-repo tasks.

func (*Task) PrimaryBaseBranch 

func (t *Task) PrimaryBaseBranch() string

PrimaryBaseBranch returns the primary repo's base-branch override, or "".

func (*Task) PushStats 

func (t *Task) PushStats(s *runtime.Stats)

PushStats records a runtime stats snapshot and notifies live subscribers.

func (*Task) RelayOffsetValue 

func (t *Task) RelayOffsetValue() int64

RelayOffsetValue returns the current relay log byte offset.

func (*Task) ReposSnapshot 

func (t *Task) ReposSnapshot() []repo.Mount

ReposSnapshot returns a copy of the task repo mounts.

func (*Task) RestoreRepositorySummary 

func (t *Task) RestoreRepositorySummary(snapshot *repo.GitSnapshot) bool

RestoreRepositorySummary installs a validated snapshot before task registration. Live and settled host observations use RecordRepositorySummary through their recording owner. It rejects stale or replaced-instance reads.

func (*Task) RuntimeConnectionTarget 

func (t *Task) RuntimeConnectionTarget() runtime.ConnectionTarget

RuntimeConnectionTarget returns the current runtime connection target.

func (*Task) RuntimeInstanceID 

func (t *Task) RuntimeInstanceID() runtime.ID

RuntimeInstanceID returns the current runtime instance ID.

func (*Task) RuntimeRepos 

func (t *Task) RuntimeRepos() []runtime.Repo

RuntimeRepos returns all repos for use with the runtime backend.

func (*Task) SeedTimelineParts 

func (t *Task) SeedTimelineParts(prefix, suffix []agent.TimedMessage)

SeedTimelineParts fills an empty task with the message history from previously saved logs, then derives session metadata, plan state, live stats and task state from it.

Seeding is for an adopted task that will resume live operation. Restored completed tasks intentionally leave the timeline empty and serve rare history reads from taskslog instead.

It panics if the task already holds messages. Seeding is one-shot initialization, not a merge: replaying a second batch onto an existing timeline would double-count cost, turns and token usage. A live timeline grows through the session message pump instead; task import runs exactly once before registration.

State inference rules (applied only for non-terminal states):

  • Current turn has unanswered AskUserQuestion → StateAsking
  • Trailing ResultMessage and an active native subagent → StateRunning
  • Trailing ResultMessage (no ask, no active subagent) → StateWaiting
  • No trailing ResultMessage → state unchanged (agent was mid-output)

Metadata-only messages (DiffStatMessage, TurnCommitSnapshotMessage, PendingUserActionMessage, NativeSubagentMessage, RawMessage) after the ResultMessage are skipped during inference. For import, the caller must handle the case where state remains StateRunning with no relay alive. The owned durable prefix and live suffix remain logically contiguous without concatenation. Both slices transfer to the task.

func (*Task) SendCompact 

func (t *Task) SendCompact(ctx context.Context, instructions string) error

SendCompact sends a compact command to the running agent without changing the task state. Returns an error if no session is active or the backend does not support compaction.

func (*Task) SendInput 

func (t *Task) SendInput(ctx context.Context, p agent.Prompt) error

SendInput sends a user message to the running agent.

Returns an error if no session is active. The error includes the task state and a SessionStatus so the caller can diagnose why the session is missing (e.g. relay died vs. never connected). The session watcher now handles dead-session detection proactively, so SendInput no longer does lazy cleanup.

func (*Task) SessionDone 

func (t *Task) SessionDone() <-chan struct{}

SessionDone returns the Done channel for the current session, or nil if no session is attached. The caller must not hold t.mu.

func (*Task) SetCIStatus 

func (t *Task) SetCIStatus(status forge.CIStatus, checks []forge.Check)

SetCIStatus updates the ciStatus and ciChecks fields under the mutex.

func (*Task) SetGitHubTokenEnabled 

func (t *Task) SetGitHubTokenEnabled(enabled bool)

SetGitHubTokenEnabled records whether this task injects a GitHub token.

func (*Task) SetPR 

func (t *Task) SetPR(owner, repoName string, pr int)

SetPR stores the forge owner, repo, and PR/MR number. Does not change task state.

func (*Task) SetPRState 

func (t *Task) SetPRState(state forge.PRState)

SetPRState updates the PR state.

func (*Task) SetRelayOffset 

func (t *Task) SetRelayOffset(offset int64)

SetRelayOffset anchors reconnect to a verified relay snapshot after import. Successful live relay appends subsequently advance this same cursor.

func (*Task) SetRepoBranch 

func (t *Task) SetRepoBranch(i int, branch string)

SetRepoBranch updates the branch for a repo by index.

func (*Task) SetRuntimeConnectionInfo 

func (t *Task) SetRuntimeConnectionInfo(id runtime.ID, target runtime.ConnectionTarget, tailscaleFQDN, tailscaleAuthURL string, vncPort int)

SetRuntimeConnectionInfo records runtime instance metadata and agent connection details.

func (*Task) SetSessionMetadata 

func (t *Task) SetSessionMetadata(m agent.SessionMetadata)

SetSessionMetadata records persisted agent session metadata.

func (*Task) SetState 

func (t *Task) SetState(s task.State)

SetState updates the state under the mutex and records the transition time.

func (*Task) SetStateAt 

func (t *Task) SetStateAt(s task.State, at time.Time)

SetStateAt updates the state under the mutex with an explicit timestamp. Used during import to preserve the original transition time.

func (*Task) SetStateIf 

func (t *Task) SetStateIf(expected, next task.State) bool

SetStateIf atomically transitions the state to next only if the current state equals expected. Returns true if the transition occurred.

func (*Task) SetStateIfAny 

func (t *Task) SetStateIfAny(next task.State, allowed ...task.State) (prev task.State, changed bool)

SetStateIfAny atomically transitions the state to next only if the current state is one of allowed. Returns the previous state and whether the transition occurred.

func (*Task) SetStateUnless 

func (t *Task) SetStateUnless(next task.State, excluded ...task.State) (prev task.State, changed bool)

SetStateUnless atomically transitions the state to next unless the current state is one of excluded, in which case it is left unchanged. Returns the previous state and whether the transition occurred. Performing the guard and the transition under a single lock closes the check-then-set race that a separate GetState/SetState pair leaves open against concurrent transitions.

func (*Task) SetSudoPassword 

func (t *Task) SetSudoPassword(password string)

SetSudoPassword caches a sudo password on the task.

func (*Task) SetTitle 

func (t *Task) SetTitle(title string)

SetTitle sets the title under the mutex, limiting it to 80 Unicode characters including a trailing ellipsis when truncated. Empty strings are ignored to preserve the prompt-fallback invariant.

func (*Task) SetTurnStartedAt 

func (t *Task) SetTurnStartedAt(at time.Time)

SetTurnStartedAt sets the turn start time if the task is currently running. Called during import to estimate when the current mid-turn started.

func (*Task) SetVNCPort 

func (t *Task) SetVNCPort(port int)

SetVNCPort records the current VNC host port.

func (*Task) Snapshot 

func (t *Task) Snapshot() Snapshot

Snapshot returns a consistent read of all volatile fields under the mutex.

func (*Task) SnapshotWithStateHistory 

func (t *Task) SnapshotWithStateHistory(after uint64) (Snapshot, uint64, []StateTransition)

SnapshotWithStateHistory returns a consistent snapshot plus the task's state transition sequence and the retained transitions newer than after, oldest first. Reading both in one critical section lets a snapshot-diffing client replay a short-lived state that the snapshot's own state already overwrote.

func (*Task) Subscribe 

func (t *Task) Subscribe(ctx context.Context) (history []TimelineMessage, live <-chan TimelineMessage, unsubFn func())

Subscribe returns a snapshot of all past messages plus a live channel for new messages. The caller must call unsubFn when done to release resources.

func (*Task) SubscribeLiveMessages 

func (t *Task) SubscribeLiveMessages(ctx context.Context) (after uint64, live <-chan TimelineMessage, unsubFn func())

SubscribeLiveMessages returns the current timeline position and sequenced messages produced after subscription. It avoids copying retained task history for consumers with another authoritative history source.

func (*Task) SubscribeRateLimits 

func (t *Task) SubscribeRateLimits(ctx context.Context) (history []*agent.RateLimitMessage, live <-chan *agent.RateLimitMessage, unsubFn func())

SubscribeRateLimits returns historical quota messages and a lossless stream of future quota messages. The caller must call unsubFn at most once when done; context cancellation automatically unsubscribes.

func (*Task) SubscribeSnapshot 

func (t *Task) SubscribeSnapshot(ctx context.Context) (history TimelineSnapshot, live <-chan TimelineMessage, unsubFn func())

SubscribeSnapshot returns a zero-copy view of past messages plus a live channel for messages appended after the snapshot. The view remains stable until history.Release or unsubFn is called. The caller must call one of them; context cancellation only unsubscribes the live channel and does not release the snapshot lease.

func (*Task) SubscribeStats 

func (t *Task) SubscribeStats(ctx context.Context) (history []runtime.Stats, live <-chan runtime.Stats, unsubFn func())

SubscribeStats returns a snapshot of the stats ring buffer and a channel that receives only live stats arriving after the snapshot. Cancellation or unsubscribing closes the channel and removes the subscriber.

func (*Task) SudoLookupState 

func (t *Task) SudoLookupState() (enabled bool, id runtime.ID, password string)

SudoLookupState returns the sudo lookup inputs and cached password.

func (*Task) TimelineID 

func (t *Task) TimelineID() string

TimelineID returns this task object's event-stream incarnation.

func (*Task) Title 

func (t *Task) Title() string

Title returns the task title under the mutex.

func (*Task) UpdateDiskUsage 

func (t *Task) UpdateDiskUsage(bytes int64)

UpdateDiskUsage records the latest writable-layer size and applies it to the most recent resource sample. Future samples inherit the value until the next disk measurement.

type TimelineMessage 

type TimelineMessage struct {
	Message    agent.Message
	Sequence   uint64
	ObservedAt time.Time
}

TimelineMessage is an immutable task message and its stable, one-based position in the task timeline.

type TimelineSnapshot 

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

TimelineSnapshot is a stable view of a task timeline at subscription time. It retains the underlying entries without copying them until Release or the subscription's unsubscribe function is called.

func (TimelineSnapshot) At 

func (s TimelineSnapshot) At(i int) TimelineMessage

At returns the indexed message with its stable one-based sequence number.

func (TimelineSnapshot) Len 

func (s TimelineSnapshot) Len() int

Len returns the number of messages retained by the snapshot.

func (TimelineSnapshot) Release 

func (s TimelineSnapshot) Release()

Release allows finalized live deltas retained for this zero-copy snapshot to be compacted. It must be called once after replay finishes.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.