ipc#

import "github.com/jbcom/radioactive-ralph/internal/ipc"

Package ipc is radioactive-ralph’s repo-service IPC layer.

The repo service listens on a local control-plane endpoint under the repo’s state directory: a Unix domain socket on macOS/Linux and a named pipe on Windows. `radioactive_ralph status`, `radioactive_ralph attach`, `radioactive_ralph stop`, and internal control-path clients exchange newline-delimited JSON messages over the same transport.

Heartbeat liveness is signalled via the repo service touching an `.alive` file every few seconds. `radioactive_ralph status` checks the file’s mtime before even attempting a socket connect — if the service crashed and left a stale socket, we want to surface the dead-service state cleanly rather than hang on a connection attempt.

Wire protocol:

Request:  {"cmd": "<verb>", "args": {...}}\n
Response: {"ok": true|false, "data": ..., "error": "..."}\n

For commands that stream (attach), the server sends N >= 0 frames of {“event”: {…}}\n followed by a terminating {“ok”: true}\n.

Index#

Constants#

Command names for the JSON-line protocol.

const (
    // v1 — observe surface.
    CmdStatus       = "status"
    CmdAttach       = "attach"
    CmdEnqueue      = "enqueue"
    CmdStop         = "stop"
    CmdReloadConfig = "reload-config"

    // v2 — drive surface (see the IPC drive-api design spec).
    CmdPlanImport    = "plan-import"
    CmdPlanSetStatus = "plan-set-status"
    CmdTaskApprove   = "task-approve"
    CmdWorkerKill    = "worker-kill"
)

Stable machine-readable error classes carried in Response.Code so a client (the GUI) can react programmatically instead of string-matching Error.

const (
    CodeUnsupportedCommand = "unsupported_command"
    CodeNotFound           = "not_found"
    CodeConflict           = "conflict"
    CodeInvalidArgs        = "invalid_args"
)

ProtoVersion is the wire protocol version this build speaks. The original read-only-TUI surface (status/attach/enqueue/stop/reload-config) is v1; the drive commands (plan-import/plan-set-status/task-approve/worker-kill) are v2. A client omitting Request.ProtoVersion is treated as v1 for back-compat.

const ProtoVersion = 2

Variables#

ErrClosed is a sentinel value; use errors.Is to match.

var ErrClosed error = closedError{}

func Dial#

func Dial(socketPath string, timeout time.Duration) (*Client, error)

Dial connects to the repo service at socketPath with the given timeout. Typical usage:

c, err := ipc.Dial(socketPath, 3*time.Second)
if err != nil { ... }
defer c.Close()
status, err := c.Status(ctx)

func IsCode#

func IsCode(err error, code string) bool

IsCode reports whether err carries the given error class. It matches any error implementing the Coded interface (Code() string) — both the client’s *CodedError (decoded from a wire Response.Code) and a handler-side coded error returned by a direct in-process call.

func NewServer#

func NewServer(opts ServerOptions) (*Server, error)

NewServer constructs a Server. It does NOT bind the socket — call Start to begin accepting connections.

func ServiceEndpoint#

func ServiceEndpoint(sessionsDir string) (endpoint, heartbeat string)

ServiceEndpoint returns the local control-plane endpoint plus its heartbeat file for one repo workspace.

On POSIX the endpoint is normally sessionsDir/service.sock. But a deeply nested sessionsDir — a long XDG/App Support path, a deep RALPH_STATE_DIR, or a macOS /var/folders/… temp root under test — can push that path past the kernel’s sun_path limit, so bind() fails with EINVAL. When that would happen we fall back to a short, collision-resistant socket path under the system temp dir keyed by a hash of sessionsDir. The heartbeat file always stays in sessionsDir (it is a plain file, not a socket, so it has no path limit) which keeps discovery/liveness colocated with the workspace.

func SocketAlive#

func SocketAlive(heartbeatPath string, maxAge time.Duration) bool

SocketAlive reports whether the heartbeat file at path was touched within maxAge. Clients (`radioactive_ralph status`) call this before attempting a socket connection so they can distinguish “service dead” from “service slow to respond.”

type AttachArgs#

AttachArgs is the client’s payload when opening an event stream via CmdAttach. ProjectID scopes the stream — the IPC connection carries no implicit project (the supervisor serves every project on one socket), so the client names it, as the drive commands do. AfterID is the client-owned resume cursor: the stream carries every event with id strictly greater than AfterID. AfterID=0 means “from the beginning” — the CLIENT, not the server, picks the live-tail cursor by first reading MaxEventID (or the backlog’s max id) and passing it here. A reconnecting client passes the highest id it has processed, resuming with no gap and no duplicate.

type AttachArgs struct {
    ProjectID string `json:"project_id"`
    AfterID   int64  `json:"after_id,omitempty"`
}

type AttachEvent#

AttachEvent is one event streamed over an Attach connection: the public, versioned shape of an events-table row. It is deliberately NOT the raw store row — Payload is the kind-specific JSON passed through verbatim, so adding a new event kind never requires a transport change. ID lets a client persist its resume cursor for reconnects.

type AttachEvent struct {
    ID         int64           `json:"id"`
    Kind       string          `json:"kind"`
    Stream     string          `json:"stream,omitempty"`
    PlanID     string          `json:"plan_id,omitempty"`
    TaskID     string          `json:"task_id,omitempty"`
    Actor      string          `json:"actor,omitempty"`
    Payload    json.RawMessage `json:"payload,omitempty"`
    OccurredAt time.Time       `json:"occurred_at"`
}

type Client#

Client wraps a single Unix-socket connection to a Ralph repo service. Clients are short-lived: construct, send one command, read reply, close. For streaming commands (attach), the client instance stays open until the server closes the stream.

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

func (*Client) Attach#

func (c *Client) Attach(ctx context.Context, args AttachArgs, fn func(json.RawMessage) error) error

Attach issues an attach request and streams raw event frames through fn until the repo service closes the stream or ctx is cancelled. The returned error is nil for a clean end-of-stream. It scopes the stream via args (the project id is required; a zero AfterID starts from the beginning). AttachEvents layers a typed decode over this; callers wanting the raw frames use Attach directly.

func (*Client) AttachEvents#

func (c *Client) AttachEvents(ctx context.Context, args AttachArgs, fn func(AttachEvent) error) error

AttachEvents is the typed convenience over Attach: it opens the stream with the given args (project scope + resume cursor) and decodes each frame into an AttachEvent before handing it to fn. Prefer this to raw Attach for the event stream; Attach stays available for callers that want the raw frames.

func (*Client) Close#

func (c *Client) Close() error

Close terminates the connection.

func (*Client) Enqueue#

func (c *Client) Enqueue(ctx context.Context, args EnqueueArgs) (EnqueueReply, error)

Enqueue pushes a task. Returns the resulting task ID (possibly a dedup hit from FTS) and whether the task was freshly inserted.

func (*Client) NegotiatedVersion#

func (c *Client) NegotiatedVersion(ctx context.Context) (int, error)

NegotiatedVersion returns the supervisor’s supported wire protocol version (from StatusReply). 0 means a pre-versioned v1 supervisor.

func (*Client) PlanImport#

func (c *Client) PlanImport(ctx context.Context, args PlanImportArgs) (PlanImportReply, error)

PlanImport imports a markdown plan and activates it, returning the created plan’s id/slug/title.

func (*Client) PlanSetStatus#

func (c *Client) PlanSetStatus(ctx context.Context, args PlanSetStatusArgs) (PlanSetStatusReply, error)

PlanSetStatus changes a plan’s lifecycle status (paused|active|abandoned).

func (*Client) ReloadConfig#

func (c *Client) ReloadConfig(ctx context.Context) error

ReloadConfig asks the repo service to re-read config.toml.

func (*Client) Status#

func (c *Client) Status(ctx context.Context) (StatusReply, error)

Status issues a status request and returns the parsed StatusReply.

func (*Client) Stop#

func (c *Client) Stop(ctx context.Context, args StopArgs) error

Stop issues a stop request. The server closes the socket after replying; expect the returned error to be ErrClosed on the next call.

func (*Client) TaskApprove#

func (c *Client) TaskApprove(ctx context.Context, args TaskApproveArgs) error

TaskApprove clears the approval gate on a ready_pending_approval task.

func (*Client) WorkerKill#

func (c *Client) WorkerKill(ctx context.Context, args WorkerKillArgs) error

WorkerKill kills a running worker via kill-and-reclaim.

type Coded#

Coded is implemented by handler errors that carry a stable machine-readable error class (Code* consts). writeResult copies it into Response.Code so the client can branch on the failure kind.

type Coded interface {
    Code() string
}

type CodedError#

CodedError wraps a !Ok Response, exposing both the human message and the stable machine-readable error class (Code* consts) so a caller (the GUI) can branch on the failure kind — e.g. treat CodeNotFound as benign. It satisfies the Coded interface.

type CodedError struct {
    Class   string
    Message string
}

func (*CodedError) Code#

func (e *CodedError) Code() string

Code returns the error class, satisfying Coded.

func (*CodedError) Error#

func (e *CodedError) Error() string

type DriveHandler#

DriveHandler is the OPTIONAL v2 drive surface. A Handler that also implements DriveHandler gains the plan-import/plan-set-status/task-approve/ worker-kill commands; one that does not still serves the v1 observe surface, and the server answers a drive command with an unsupported_command response. Keeping it a separate interface means existing v1 Handler implementations (and their test doubles) compile unchanged.

type DriveHandler interface {
    // HandlePlanImport creates + activates a plan from markdown.
    HandlePlanImport(ctx context.Context, args PlanImportArgs) (PlanImportReply, error)
    // HandlePlanSetStatus changes a plan's lifecycle status (validated).
    HandlePlanSetStatus(ctx context.Context, args PlanSetStatusArgs) (PlanSetStatusReply, error)
    // HandleTaskApprove clears the approval gate on a ready_pending_approval task.
    HandleTaskApprove(ctx context.Context, args TaskApproveArgs) error
    // HandleWorkerKill kills a running worker via kill-and-reclaim.
    HandleWorkerKill(ctx context.Context, args WorkerKillArgs) error
}

type EnqueueArgs#

EnqueueArgs is the client’s payload when pushing work via CmdEnqueue.

type EnqueueArgs struct {
    TaskID      string `json:"task_id"` // optional; service generates UUID if empty
    Description string `json:"description"`
    Priority    int    `json:"priority,omitempty"`
}

type EnqueueReply#

EnqueueReply tells the client whether a new task was created or a duplicate was collapsed (via FTS dedup in the db layer).

type EnqueueReply struct {
    TaskID   string `json:"task_id"`
    Inserted bool   `json:"inserted"` // false means FTS found a duplicate
}

type Handler#

Handler handles a single client request. Attach streams events by calling emit repeatedly; other commands return (reply, nil) and the server transmits a single Response frame.

type Handler interface {
    // HandleStatus returns the current repo-service status.
    HandleStatus(ctx context.Context) (StatusReply, error)

    // HandleEnqueue queues a new task, returning the task ID and whether
    // it was a fresh insert or a dedup hit.
    HandleEnqueue(ctx context.Context, args EnqueueArgs) (EnqueueReply, error)

    // HandleStop signals the repo service to shut down. The server closes
    // the IPC socket after sending the response.
    HandleStop(ctx context.Context, args StopArgs) error

    // HandleReloadConfig asks the repo service to re-read config.toml.
    HandleReloadConfig(ctx context.Context) error

    // HandleAttach streams events to the client until either the
    // service exits or the client disconnects. args.AfterID is the resume
    // cursor — the handler emits events with id greater than it (0 means
    // start from the live tail). The implementation should return when ctx
    // is cancelled.
    HandleAttach(ctx context.Context, args AttachArgs, emit func(json.RawMessage) error) error
}

type OKReply#

OKReply is the trivial success payload for drive commands that only need to confirm the action landed.

type OKReply struct {
    OK bool `json:"ok"`
}

type PlanImportArgs#

PlanImportArgs imports a markdown plan and activates it (CmdPlanImport). The server runs the same CreatePlan + activate logic the `plan import` CLI does, so the GUI needn’t open the DB itself and there is one writer of record.

type PlanImportArgs struct {
    Markdown string `json:"markdown"`
    Slug     string `json:"slug,omitempty"`  // optional; derived from title if empty
    Title    string `json:"title,omitempty"` // optional; derived from first heading/filename if empty
    Project  string `json:"project"`         // project id the plan belongs to
}

type PlanImportReply#

PlanImportReply reports the created plan.

type PlanImportReply struct {
    PlanID string `json:"plan_id"`
    Slug   string `json:"slug"`
    Title  string `json:"title"`
}

type PlanSetStatusArgs#

PlanSetStatusArgs changes a plan’s lifecycle status (CmdPlanSetStatus), e.g. pause/resume/abandon. The server validates the transition.

type PlanSetStatusArgs struct {
    PlanID string `json:"plan_id"`
    Status string `json:"status"` // paused|active|abandoned
}

type PlanSetStatusReply#

PlanSetStatusReply echoes the applied status.

type PlanSetStatusReply struct {
    PlanID string `json:"plan_id"`
    Status string `json:"status"`
}

type Request#

Request is a single command from a client to the repo service.

type Request struct {
    Cmd  string          `json:"cmd"`
    Args json.RawMessage `json:"args,omitempty"`
    // ProtoVersion is the wire version the client speaks. 0 (omitted) means a
    // pre-versioned v1 client (the current TUI), handled for back-compat.
    ProtoVersion int `json:"proto_version,omitempty"`
}

type Response#

Response is the single-shot reply shape. For streaming commands the server sends multiple Event frames followed by a final Response with Ok=true; mid-stream errors send a Response with Ok=false.

type Response struct {
    Ok    bool            `json:"ok"`
    Data  json.RawMessage `json:"data,omitempty"`
    Error string          `json:"error,omitempty"`
    // Code is a stable machine-readable error class (Code* consts) set on
    // !Ok responses where the client may want to branch on the failure kind.
    Code string `json:"code,omitempty"`
}

type Server#

Server is the repo-service IPC server. One instance per repo service.

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

func (*Server) Start#

func (s *Server) Start() error

Start binds the socket and begins accepting connections in a background goroutine. Safe to call once. Returns the listener error if bind fails. The heartbeat interval comes from ServerOptions.HeartbeatInterval (set at NewServer), not a parameter here — a single source of truth.

func (*Server) Stop#

func (s *Server) Stop() error

Stop shuts the server down and waits for goroutines to exit.

type ServerOptions#

ServerOptions configures a Server.

type ServerOptions struct {
    // SocketPath is the local endpoint path to bind. Typically
    // state/<repo>/sessions/service.sock on POSIX hosts.
    SocketPath string

    // HeartbeatPath is the file whose mtime is bumped every
    // HeartbeatInterval by the server. Typically the repo-service
    // heartbeat file next to the endpoint metadata.
    HeartbeatPath string

    // HeartbeatInterval controls how often we refresh the heartbeat.
    // Defaults to 10s when zero.
    HeartbeatInterval time.Duration

    // Handler satisfies the IPC Handler interface.
    Handler Handler

    // Logger receives info/warn/error messages. Defaults to a no-op
    // logger when nil.
    Logger *slog.Logger
}

type StatusReply#

StatusReply is the data payload for CmdStatus responses.

type StatusReply struct {
    // ProtoVersion is the supervisor's supported wire version, so a client
    // can detect drive-command availability without trial-and-error.
    ProtoVersion  int             `json:"proto_version,omitempty"`
    RepoPath      string          `json:"repo_path"`
    PID           int             `json:"pid"`
    Uptime        time.Duration   `json:"uptime_ns"`
    ActiveWorkers int             `json:"active_workers"`
    ReadyTasks    int             `json:"ready_tasks"`
    ApprovalTasks int             `json:"approval_tasks"`
    BlockedTasks  int             `json:"blocked_tasks"`
    RunningTasks  int             `json:"running_tasks"`
    FailedTasks   int             `json:"failed_tasks"`
    ActivePlans   int             `json:"active_plans"`
    Workers       []WorkerSummary `json:"workers,omitempty"`
    LastEventAt   time.Time       `json:"last_event_at,omitempty"`
    HeartbeatAge  time.Duration   `json:"heartbeat_age_ns,omitempty"`
}

type StopArgs#

StopArgs controls the termination mode for CmdStop.

type StopArgs struct {
    Graceful bool          `json:"graceful"`             // wait for in-flight sessions to finish cleanly
    Timeout  time.Duration `json:"timeout_ns,omitempty"` // overrides default if >0
}

type StreamEvent#

StreamEvent is one frame emitted during a streaming command (e.g. attach).

type StreamEvent struct {
    Event json.RawMessage `json:"event"`
}

type TaskApproveArgs#

TaskApproveArgs clears the approval gate on a ready_pending_approval task (CmdTaskApprove), transitioning it to ready so dispatch can pick it up.

type TaskApproveArgs struct {
    PlanID string `json:"plan_id"`
    TaskID string `json:"task_id"`
}

type WorkerKillArgs#

WorkerKillArgs kills a running worker (CmdWorkerKill) via the same kill-and-reclaim path a watchdog kill uses, so the task returns to ready.

type WorkerKillArgs struct {
    WorkerID string `json:"worker_id"`
}

type WorkerSummary#

WorkerSummary is the runtime-facing status for one in-flight worker.

type WorkerSummary struct {
    // WorkerID is the store worker-row id — the value a client passes to the
    // worker-kill drive command to target THIS worker. Distinct from any
    // provider-session id.
    WorkerID          string `json:"worker_id"`
    PlanID            string `json:"plan_id"`
    TaskID            string `json:"task_id"`
    Provider          string `json:"provider,omitempty"`
    ProviderSessionID string `json:"provider_session_id,omitempty"`
}

Generated by gomarkdoc