container

package
v0.0.0-...-a89d666 Latest Latest
Warning

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

 Go to latest Published: Oct 4, 2026 License: MIT Imports: 44 Imported by: 0

Documentation

Overview

Package container manages Docker container lifecycle for task execution.

Index

Constants

View Source 
const (
	// LabelRunner marks containers managed by contextmatrix-runner.
	LabelRunner = "contextmatrix.runner"
	// LabelCardID stores the card ID on the container.
	LabelCardID = "contextmatrix.card_id"
	// LabelProject stores the project name on the container.
	LabelProject = "contextmatrix.project"
	// LabelSessionID stores the chat session ID on chat-mode containers.
	LabelSessionID = "contextmatrix.session_id"
)

Variables

This section is empty.

Functions

func IsValidStaticSecretByte 

func IsValidStaticSecretByte(b byte) bool

IsValidStaticSecretByte reports whether b is acceptable inside the shell-sourced /run/cm-secrets/env value of a static secret. Real OAuth tokens and API keys live in the printable-ASCII range; rejecting everything outside that range catches the dangerous bytes (NUL, \n, \r, \t, escape sequences, raw UTF-8 multi-byte starts) without over-fitting to current token shapes.

func MonitorDockerd 

func MonitorDockerd(ctx context.Context, docker DockerClient, logger *slog.Logger)

MonitorDockerd runs the health-probe loop without a metrics bundle. Used by callers (typically tests) that have no Prometheus registry wired. Production callers should prefer MonitorDockerdWithMetrics so the per-iteration recover increments cmr_panic_recovered_total.

The function takes a DockerClient (not a *RealDockerClient) so tests can swap in a MockDockerClient with a scripted PingFn. A nil logger is accepted and replaced with slog.Default() — this is defensive only; production wiring should always pass an explicit logger so the restart reason is logged with the runner's structured fields.

func MonitorDockerdWithMetrics 

func MonitorDockerdWithMetrics(ctx context.Context, docker DockerClient, logger *slog.Logger, mx *metrics.Metrics)

MonitorDockerdWithMetrics runs until ctx is cancelled, pinging dockerd every healthProbeInterval. On healthFailureLimit consecutive failures it calls exitFn(1) so systemd (or whatever supervisor the operator deploys) can restart the runner and recover from pathological SDK states the docker SDK's internal reconnect logic does not handle.

A panic inside pingOnce (typically a bug in the Docker SDK or a corrupted internal state on a third-party reconnect attempt) would otherwise unwind the whole goroutine; main.go's outer recover-and-respawn loop would catch that but lose a short window of monitoring. The per-iteration recover here keeps the same goroutine alive so the failure counter and ticker cadence are preserved across pathological probes. mx may be nil for callers that have no Prometheus registry wired (tests).

Types

type ChatResume 

type ChatResume struct {
	Turns   []ChatResumeTurn
	Clipped bool
	OrigSeq int64
}

ChatResume mirrors the wire shape of the rehydration payload. Defined here (rather than re-using webhook.ChatResumeContext) so the container package has no upward dependency on webhook types.

type ChatResumeTurn 

type ChatResumeTurn struct {
	Seq     int64  `json:"seq"`
	Role    string `json:"role"`
	Content string `json:"content"`
}

ChatResumeTurn is one filtered, possibly summarized transcript entry.

type DockerClient 

type DockerClient interface {
	// Ping verifies that the Docker daemon is reachable. Used by the startup
	// preflight and the background health monitor. The long-lived
	// RealDockerClient is expected to reconnect automatically across most
	// dockerd restarts via the docker SDK's internal connection handling;
	// the dockerd health monitor's os.Exit(1) on 3 consecutive failures is
	// the escape hatch for cases where the SDK does not recover on its own.
	Ping(ctx context.Context) error
	ImagePull(ctx context.Context, ref string, options image.PullOptions) (io.ReadCloser, error)
	// ImageInspect returns image metadata. The manager only uses this to
	// confirm presence (nil error = image exists locally); the full
	// response is not consumed today but is threaded through the interface
	// so tests can assert inspection calls happened.
	ImageInspect(ctx context.Context, imageID string) (image.InspectResponse, error)
	ContainerCreate(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, networkConfig *network.NetworkingConfig, platform *ocispec.Platform, name string) (container.CreateResponse, error)
	ContainerStart(ctx context.Context, containerID string, options container.StartOptions) error
	ContainerWait(ctx context.Context, containerID string, condition container.WaitCondition) (<-chan container.WaitResponse, <-chan error)
	ContainerStop(ctx context.Context, containerID string, options container.StopOptions) error
	ContainerRemove(ctx context.Context, containerID string, options container.RemoveOptions) error
	ContainerLogs(ctx context.Context, containerID string, options container.LogsOptions) (io.ReadCloser, error)
	ContainerList(ctx context.Context, options container.ListOptions) ([]DockerContainer, error)
	ContainerAttach(ctx context.Context, containerID string, options container.AttachOptions) (*HijackedResponse, error)
	// ImagesPrune asks dockerd to delete dangling/unused images matching the
	// given filter args. Used by the periodic maintenance loop to keep local
	// image cache bounded — without this, the host image store grows
	// unbounded across worker-image upgrades.
	ImagesPrune(ctx context.Context, pruneFilter filters.Args) (image.PruneReport, error)
	Close() error
}

DockerClient abstracts the Docker SDK methods used by the manager. This interface enables testing with mocks.

type DockerContainer 

type DockerContainer struct {
	ID      string
	Names   []string
	State   string
	Created int64 // Unix seconds; matches the Docker SDK's types.Container.Created
	Labels  map[string]string
}

DockerContainer is a simplified container info struct used by ContainerList.

State / Created / Names are populated by RealDockerClient.ContainerList and consumed by the /containers webhook endpoint so CM's reconcile sweep can cross-reference live Docker state against the card store without relying on the runner's in-memory tracker (which can diverge from Docker on a cleanup failure — the exact class of bug the Docker-authoritative sweep fixes).

type HijackedResponse 

type HijackedResponse struct {
	// Conn is the write side of the hijacked stdin connection.
	Conn io.WriteCloser
	// contains filtered or unexported fields
}

HijackedResponse wraps the Docker SDK's hijacked connection and exposes only the write side. This avoids leaking the SDK type through the DockerClient interface, keeping it fully mockable.

func (*HijackedResponse) Close 

func (h *HijackedResponse) Close()

Close releases the underlying network connection.

type ManagedContainer 

type ManagedContainer struct {
	ContainerID   string
	ContainerName string
	CardID        string
	SessionID     string
	Project       string
	State         string
	StartedAt     time.Time
	Tracked       bool
}

ManagedContainer describes a Docker container labeled as runner-managed. It is the ground-truth unit consumed by CM's reconcile sweep: a container is listed here iff docker ps says so, regardless of whether the runner's in-memory tracker still knows about it. That divergence is the failure mode the Docker-authoritative sweep is designed to catch.

type Manager 

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

Manager handles the lifecycle of Docker containers for task execution.

func NewManager 

func NewManager(
	docker DockerClient,
	tracker *tracker.Tracker,
	cb *callback.Client,
	token githubauth.TokenGenerator,
	broadcaster *logbroadcast.Broadcaster,
	cfg *config.Config,
	logger *slog.Logger,
) *Manager

NewManager creates a container manager.

token (githubauth.TokenGenerator) is expected to be non-nil for any deployment that actually spawns containers — every spawn-path branch (card mode startContainer, chat mode StartChat / MintChatGitToken, and the background tokenRefresher) calls GenerateToken conditionally. Production wiring in cmd/contextmatrix-runner/main.go always passes a real provider.

A nil token is permitted here so that test code which only exercises non-spawn methods (Kill, CleanupOrphans, ListManaged, PruneImages, etc.) can construct a Manager without spinning up a fake GitHub token server. The four spawn-path call sites (startContainer, buildSecretDelivery, StartTokenRefresher, MintChatGitToken) each guard with a nil-check or a TaskSkillsDir-guard that skips the mint when the token has no consumer.

cb and broadcaster may also be nil; the manager guards those at every call site.

func (*Manager) AttachChatStdin 

func (m *Manager) AttachChatStdin(ctx context.Context, sessionID, containerID string) error

AttachChatStdin opens the chat container's stdin and registers the writer with the tracker under the given sessionID. Without this, /message webhooks cannot route user turns into the running claude process. Must be called AFTER tracker.AddChat for the session.

func (*Manager) CleanupOrphans 

func (m *Manager) CleanupOrphans(ctx context.Context) error

CleanupOrphans removes any leftover containers from a previous runner crash. A container is "orphan" iff it is labeled LabelRunner=true in Docker AND has no corresponding entry in the in-memory tracker — i.e. the current runner process does not know about it. Containers that are actively tracked (a card is assigned, the runner is still managing them) are skipped so the periodic maintenance sweep does not kill live work.

Only Remove failures are returned as errors. A Stop failure that ended in a successful Remove is logged as a warning but not surfaced to the caller — the container is gone, which is the only outcome that matters. Returning a Stop failure as a hard error would mislead callers into thinking cleanup did not complete, even when every orphan was ultimately destroyed.

func (*Manager) DeleteChatCleanup 

func (m *Manager) DeleteChatCleanup(containerID string)

DeleteChatCleanup pops the cleanup state stashed by StartChat for the given container ID and removes the chat resume dir. Called by webhook rollback paths that abort BEFORE invoking WaitAndCleanupChat — in those paths the wait goroutine is not running, so no one else will perform the cleanup. Safe to call on an unknown ID: returns without touching the filesystem. The shared secrets dir is NOT removed — it is owned by the tokenRefresher.

func (*Manager) ForceKillContainer 

func (m *Manager) ForceKillContainer(ctx context.Context, containerID string) error

ForceKillContainer is the shutdown-path backstop: stop + force-remove a container directly by ID, ignoring the tracker. Used by the main shutdown sequence's force-cleanup phase, after the normal Kill + mgr.Wait has already timed out. The caller must pass a bounded ctx.

func (*Manager) ForceRemoveByLabels 

func (m *Manager) ForceRemoveByLabels(ctx context.Context, project, cardID string) (int, error)

ForceRemoveByLabels is the /kill fallback path: when the tracker has no entry for (project, cardID) but Docker still holds a labeled container, we bypass the tracker-driven cancel flow entirely and go straight to docker rm -f. The only sane way to get here is tracker/Docker divergence (a prior cleanup returned early with a logged warning before removal succeeded) — in which case every additional layer that "properly" cancels the missing tracker entry is a no-op, and the container leaks to the 2h container_timeout unless we reach past them.

Returns the number of containers removed. An error from any single removal is joined into the final error but does not stop the sweep over the rest of the matches — removing as many as possible still beats leaving them all running.

func (*Manager) InitSharedSecrets 

func (m *Manager) InitSharedSecrets() error

func (*Manager) Kill 

func (m *Manager) Kill(project, cardID string) error

Kill stops and removes a specific container by project and card ID.

func (*Manager) KillChat 

func (m *Manager) KillChat(ctx context.Context, sessionID string) error

KillChat is the chat-mode equivalent of Kill. It closes the container's stdin (so claude has a chance to flush its final stream-json batch) and then calls Stop, giving the container its normal SIGTERM grace period before SIGKILL. If no stdin has been attached yet (the window between AddChat and SetStdinChat) the close is skipped and Stop is still called.

func (*Manager) ListManaged 

func (m *Manager) ListManaged(ctx context.Context) ([]ManagedContainer, error)

ListManaged returns every Docker container labeled LabelRunner=true, regardless of running/exited state. Tracked reflects whether the in-memory tracker currently has a matching entry; consumers can use the field to detect tracker/Docker divergence without needing a second round-trip.

Two kinds are surfaced: card-mode containers (LabelCardID + LabelProject) and chat-mode containers (LabelSessionID, LabelProject optional — a global chat may have no project label). Containers with neither identifier set are skipped — they are neither reachable via /kill (which routes by labels) nor the sweep's responsibility. Such containers still exist in Docker and are caught by CleanupOrphans on the next maintenance tick.

func (*Manager) MintChatGitToken 

func (m *Manager) MintChatGitToken(ctx context.Context) string

MintChatGitToken mints a GitHub installation token for the runner's own taskSkillsMount git pull. Claude auth and CM_GIT_TOKEN for the container are owned by the tokenRefresher and reach the worker via the shared secrets dir — this token must not be forwarded to the container.

When cfg.TaskSkillsDir is empty the token has no consumer (the skills mount is the only caller that uses it), so we skip the GitHub round-trip entirely. Card-mode startContainer skips the mint in the same case; this guard keeps the two modes symmetric so a deployment without a skills dir pays zero GitHub API calls on /chat/start.

func (*Manager) PruneImages 

func (m *Manager) PruneImages(ctx context.Context) error

PruneImages asks dockerd to delete dangling/unused images older than imagePruneMaxAge. Called on each tick of the maintenance loop. Returning an error does not stop the loop — the caller logs and continues. Called once per maintenance tick from the reconcile-and-prune loop.

func (*Manager) Run 

func (m *Manager) Run(ctx context.Context, payload RunConfig)

Run launches the full container lifecycle for a triggered task in a goroutine. Use Wait to block until all launched goroutines have finished.

func (*Manager) StartChat 

func (m *Manager) StartChat(ctx context.Context, opts StartChatOpts) (string, error)

StartChat creates and starts a chat-mode worker container. The container runs claude in stream-json mode; the entrypoint dispatches on CM_CHAT_SESSION. Returns the container ID. The caller is responsible for (1) registering the container in the tracker via AddChat, and (2) attaching stdin via AttachChatStdin so /message webhooks can route user turns in.

func (*Manager) StartTokenRefresher 

func (m *Manager) StartTokenRefresher(ctx context.Context) error

StartTokenRefresher initialises the shared secrets directory via InitSharedSecrets, performs the first token mint synchronously, and then spawns a background goroutine that re-mints on expiry. The initial mint is fail-closed: if it fails, StartTokenRefresher returns the error immediately and the caller should treat it as fatal at boot time. Subsequent mint failures in the goroutine loop are retried with exponential backoff and do not propagate. In env-var fallback mode (cfg.SecretsDir empty), InitSharedSecrets is a no-op, the initial mint is skipped, and the spawned goroutine logs and returns immediately.

Returns an error if m.token is nil (a deployment that bind-mounts a shared secrets dir but has no token provider configured is a misconfig).

func (*Manager) Stop 

func (m *Manager) Stop(ctx context.Context, containerID string) error

Stop stops and force-removes a container by ID. It is a thin wrapper used for rollback on /chat/start when tracker registration fails after the container is already running. Errors are collected but not fatal — the caller logs and moves on.

func (*Manager) StreamChatLogs 

func (m *Manager) StreamChatLogs(ctx context.Context, sessionID, containerID, project string)

StreamChatLogs follows the chat container's stdout/stderr in a background goroutine and republishes claude stream-json events as LogEntry values on the broadcaster, keyed by sessionID. The goroutine exits when ctx is cancelled or the log reader EOFs (container exit). Caller must hold a reference to ctx and cancel it on session end or shutdown; the outer goroutine plus each of its three children (stdcopy, stderr scanner, and logparser) are tracked by m.wg so Wait() drains them.

Secrets stashed by StartChat under containerID are popped here and used to construct a per-container Redactor. Without redaction, CM_MCP_API_KEY and any env-var-mode secrets could leak into the /logs SSE stream and slog output via container stdout/stderr. Each inner goroutine is wrapped in a recover() so a malformed stream-json or stdcopy frame in one chat container does not crash the whole runner.

func (*Manager) SweepStaleChatResumeDirs 

func (m *Manager) SweepStaleChatResumeDirs(maxAge time.Duration)

SweepStaleChatResumeDirs removes cmr-chat-resume-* host directories that are older than maxAge. These directories are created by prepareChatResume and cleaned up by WaitAndCleanupChat; any that survive are crash-leak artefacts. Candidates are searched under SecretsDir (primary) and os.TempDir() (dev fallback). Per-entry removal errors are logged and silently swallowed — this is a best-effort janitor.

func (*Manager) Wait 

func (m *Manager) Wait()

Wait blocks until all container goroutines launched by Run have finished.

func (*Manager) WaitAndCleanupChat 

func (m *Manager) WaitAndCleanupChat(sessionID, containerID, project string)

WaitAndCleanupChat launches a wg-tracked goroutine that waits for the chat container to exit (any reason — claude crash, OOM, external kill, or a /chat/end-triggered Stop) and then unconditionally removes the tracker entry and force-removes the container.

/chat/end already performs this teardown explicitly, so the goroutine races it; that's fine — every step is idempotent (RemoveChat tolerates missing entries; ContainerRemove on a removed container is logged and swallowed). The point of this goroutine is to catch the *implicit* exit paths where /chat/end is never called: claude crash, OOM, exec failure, or `docker kill` from outside the runner. Without it those paths would leak the tracker entry until the runner restarts.

Uses context.Background() (optionally wrapped in a WithTimeout derived from m.cfg.ChatContainerTimeout) so the wait survives the request ctx that kicked off /chat/start. ChatContainerTimeout is an opt-in, generous lifetime backstop (default 0 = disabled): when set, a chat container that never exits — a wedged claude process or a hung dockerd wait, with /chat/end never arriving — is force-stopped once the deadline fires instead of leaking the goroutine and its tracker slot forever. It is not an idle-output watchdog and will not kill a legitimately-idle human session. The associated log-stream goroutine (from StreamChatLogs) exits on its own when ContainerLogs EOFs on container removal, so no explicit cancel is required here.

The webhook handler owns the lifecycle: StartChat returns the container ID after stashing the delivery handles in m.chatCleanup, and the handler invokes WaitAndCleanupChat only after registering the tracker entry. The secret / resume cleanup handles are popped from the chatCleanup map so callers don't need to thread them through the signature.

func (*Manager) WithMetrics 

func (m *Manager) WithMetrics(mx *metrics.Metrics) *Manager

WithMetrics attaches a metrics bundle to the manager. A nil bundle disables metric observation (useful in tests that do not care about Prometheus).

func (*Manager) WorkerImage 

func (m *Manager) WorkerImage() string

WorkerImage returns the configured base image name. Used by the webhook handler to record which image a chat container was started from.

type RealDockerClient 

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

RealDockerClient wraps the Docker SDK client.

func NewRealDockerClient 

func NewRealDockerClient() (*RealDockerClient, error)

NewRealDockerClient creates a Docker client from the environment.

func (*RealDockerClient) Close 

func (c *RealDockerClient) Close() error

func (*RealDockerClient) ContainerAttach 

func (c *RealDockerClient) ContainerAttach(ctx context.Context, containerID string, options container.AttachOptions) (*HijackedResponse, error)

func (*RealDockerClient) ContainerCreate 

func (c *RealDockerClient) ContainerCreate(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, networkConfig *network.NetworkingConfig, platform *ocispec.Platform, name string) (container.CreateResponse, error)

func (*RealDockerClient) ContainerList 

func (c *RealDockerClient) ContainerList(ctx context.Context, options container.ListOptions) ([]DockerContainer, error)

func (*RealDockerClient) ContainerLogs 

func (c *RealDockerClient) ContainerLogs(ctx context.Context, containerID string, options container.LogsOptions) (io.ReadCloser, error)

func (*RealDockerClient) ContainerRemove 

func (c *RealDockerClient) ContainerRemove(ctx context.Context, containerID string, options container.RemoveOptions) error

func (*RealDockerClient) ContainerStart 

func (c *RealDockerClient) ContainerStart(ctx context.Context, containerID string, options container.StartOptions) error

func (*RealDockerClient) ContainerStop 

func (c *RealDockerClient) ContainerStop(ctx context.Context, containerID string, options container.StopOptions) error

func (*RealDockerClient) ContainerWait 

func (c *RealDockerClient) ContainerWait(ctx context.Context, containerID string, condition container.WaitCondition) (<-chan container.WaitResponse, <-chan error)

func (*RealDockerClient) ImageInspect 

func (c *RealDockerClient) ImageInspect(ctx context.Context, imageID string) (image.InspectResponse, error)

func (*RealDockerClient) ImagePull 

func (c *RealDockerClient) ImagePull(ctx context.Context, ref string, options image.PullOptions) (io.ReadCloser, error)

func (*RealDockerClient) ImagesPrune 

func (c *RealDockerClient) ImagesPrune(ctx context.Context, pruneFilter filters.Args) (image.PruneReport, error)

func (*RealDockerClient) Ping 

func (c *RealDockerClient) Ping(ctx context.Context) error

Ping checks whether the Docker daemon is reachable. It returns nil on success and a non-nil error on connectivity or daemon-internal failures.

Reconnect contract: the long-lived *client.Client created in NewRealDockerClient is expected to re-establish its underlying connection across most dockerd restarts. That behavior comes from the docker SDK, not this wrapper. The dockerd health monitor (see cmd/contextmatrix-runner) exits the process after 3 consecutive failures, which delegates recovery from pathological SDK states (e.g. a stuck keepalive) to systemd restart.

type RunConfig 

type RunConfig struct {
	CardID        string
	Project       string
	RepoURL       string
	MCPURL        string
	MCPAPIKey     string
	BaseBranch    string
	RunnerImage   string
	Interactive   bool
	Model         string
	CorrelationID string
	// TaskSkills is the optional set of skill names to activate in the container.
	// nil means "no constraint" (all skills); non-nil (even empty) means the
	// set is explicit. The entrypoint uses CM_TASK_SKILLS_SET to distinguish
	// the two cases.
	TaskSkills *[]string
}

RunConfig holds the parameters needed to start a container. This mirrors the webhook TriggerPayload but avoids an import cycle.

type StartChatOpts 

type StartChatOpts struct {
	// SessionID is required; the container is labelled and the CM_CHAT_SESSION
	// env var is set to this value.
	SessionID string
	// Project is optional; sets CM_CHAT_PROJECT when non-empty.
	Project string
	// RepoURL is optional; sets CM_CHAT_REPO_URL when non-empty.
	RepoURL string
	// MCPURL is the URL the in-container claude uses to reach CM's MCP
	// endpoint (e.g. http://host.docker.internal:8080/mcp). Sets
	// CM_MCP_URL. Without this, the entrypoint skips the MCP merge and
	// chat sessions cannot call ContextMatrix tools.
	MCPURL string
	// MCPAPIKey authenticates MCP calls. Sets CM_MCP_API_KEY. Empty is
	// permitted when CM's MCP listener has no auth (loopback dev mode).
	MCPAPIKey string
	// GitToken is the GitHub installation token used solely for the runner's
	// own taskSkillsMount git pull and never reaches the container env or
	// shared secrets file.
	GitToken string
	// Model is the Claude model the chat container should run. Sets
	// CM_ORCHESTRATOR_MODEL; the entrypoint passes this to `claude --model`.
	// Empty falls back to the entrypoint default (claude-sonnet-4-6).
	Model string
	// Resume, when non-nil, makes the runner write /run/cm-chat/resume.jsonl
	// and /run/cm-chat/resume.meta.json into the container and set
	// CM_CHAT_RESUME=1. The entrypoint switches to the rehydration prompt
	// branch when this flag is present.
	Resume *ChatResume
	// TaskSkills mirrors RunConfig.TaskSkills for chat-mode workers. nil =
	// no constraint (entrypoint copies the full set from /host-skills);
	// non-nil (even empty) = explicit selection via CM_TASK_SKILLS_SET +
	// CM_TASK_SKILLS. The /host-skills bind-mount is added whenever
	// cfg.TaskSkillsDir is set, regardless of this field.
	TaskSkills *[]string
}

StartChatOpts configures a chat-mode container spawn.

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.