dispatch

package
v0.0.3 Latest Latest
Warning

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

 Go to latest Published: May 6, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

Package dispatch is the execution-layer dispatcher (PR-42). It picks pending tasks off the store, claims their soft locks, spawns one cmux workspace per task, and sends a Claude prompt that combines the base execution skill, the source-specific skill (if any), and the task body. See docs/requirement.md "Execution（実行）— ディスパッチャ詳細".

The package is intentionally split:

  • prompt.go — pure prompt assembly (BuildPrompt). No I/O so it can be unit-tested without spinning up cmux / sqlite.
  • lockkey.go — notes.lock_hint -> [execution.lock_keys] resolver.
  • dispatch.go — Run / Dispatcher: ties the cmux client + store repo together with the lock-key resolver and prompt builder.

Index

Constants

This section is empty.

Variables

View Source 
var ErrInvalidConfig = errors.New("dispatch: missing required option")

ErrInvalidConfig signals a missing required Option at construction.

View Source 
var ErrNotPending = errors.New("dispatch: task is not pending")

ErrNotPending signals that `marunage dispatch <id>` was asked to dispatch a row that is not currently pending. Distinct from store.ErrNotFound so the CLI can print a precise diagnostic.

Functions

func BuildPrompt 

func BuildPrompt(in PromptInputs) string

BuildPrompt concatenates (Base, SourceSpecific, Triage, Task, Sentinel) in that fixed order. Empty sections drop out cleanly.

User-derived fields (Source, ExternalID, ExternalURL, Title, Body) go through fenceEscape so a malicious payload cannot splice a forged fence-close + override into the prompt. requirement.md L29 invariant #2 ("No silent execution") is the upstream policy this satisfies: the receiving Claude session can refuse to follow instructions that originate from inside a `<<body>>` / `<<title>>` fence.

The Send wrapper in internal/cmux collapses any embedded \r\n run into a single space before handing the payload to cmux; preserving the original line breaks here keeps the prompt readable when the caller inspects it via `marunage show <id>` or the Web UI.

func ResolveLockKey 

func ResolveLockKey(rules map[string]string, notes string) (string, error)

ResolveLockKey looks up the soft-lock key the dispatcher should AcquireLock on for a row, by extracting notes."lock_hint" and matching it against the regex map loaded from [execution.lock_keys] in config.toml. Returns "" when no rule matches; callers skip AcquireLock in that case.

rules has the shape {regex -> lock_key}. The resolver sorts the regex keys lexicographically before iterating so that two dispatcher runs against the same row always pick the same lock_key — Go map iteration order is intentionally randomised, and a non-deterministic resolver would let two parallel marunage instances race against each other for the same row's lock.

Errors:

  • invalid notes JSON: returned as-is so a Discovery plugin bug fails loud rather than silently dropping the lock_hint.
  • invalid regex in rules: returned as-is, naming the offending pattern so the user can fix config.toml.

Tolerances (intentionally not errors):

  • empty notes ("" or NULL on the wire) -> ""
  • notes is not a JSON object (string, array, ...) -> ""
  • notes is an object without a lock_hint key -> ""
  • lock_hint is null or "" -> ""

Types

type Dispatcher 

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

Dispatcher ties the cmux client + store repo together with the lock-key resolver and prompt builder.

func New 

func New(opts ...Option) (*Dispatcher, error)

New builds a Dispatcher. Required: WithStore, WithCmux, WithBaseSkill, WithClaudeCommand. Returns ErrInvalidConfig naming the missing field so a buggy CLI wiring fails loud at startup.

func (*Dispatcher) HandlePermissionRequest 

func (d *Dispatcher) HandlePermissionRequest(ctx context.Context, taskID int64, tool, args string) (PermissionDecision, error)

HandlePermissionRequest is the dispatcher-side handler for one Claude tool-permission request. The caller (a future cmux/MCP shim) supplies the (tool, args) pair Claude wants to invoke; this method consults the configured PermissionMatcher and on_unknown_permission policy, mutates the row when the policy demands it, records an audit entry, and returns the verdict.

Decision matrix: 

matcher.Allow(tool, args) == true       -> PermissionAllow (no row mutation)
matcher == nil                          -> PermissionAsk   (safe default)
matcher denies + policy "escalate"      -> EscalateToHuman + dispatch.escalate audit -> PermissionEscalate
matcher denies + policy "fail"          -> MarkFailedWithReason + dispatch.fail audit -> PermissionFail
matcher denies + policy "retry" or ""   -> PermissionAsk   (caller re-prompts)

The reason string written to audit / judgment_reason runs through logging.Redact so a tool args payload that happens to carry a Bearer header / API key does not pin the secret into either sink.

func (*Dispatcher) Run 

func (d *Dispatcher) Run(ctx context.Context, opts RunOptions) error

Run picks pending rows in dispatch order and pushes each into a fresh cmux workspace. Per-row failures are isolated so one stuck row does not poison the whole batch (docs/requirement.md "他のソースの discovery / 既存タスクの dispatch は継続（巻き込み故障させない）"). The returned error is reserved for catastrophic failures (e.g. Store.List itself blew up); per-row failures are written into the row's status / judgment_reason and Run returns nil.

type Option 

type Option func(*Dispatcher)

Option mutates Dispatcher construction.

func WithAllowedCwdPrefixes 

func WithAllowedCwdPrefixes(prefixes []string) Option

WithAllowedCwdPrefixes installs the cfg.Execution.AllowedCwdPrefixes allowlist. A row whose CWD does not start with any listed prefix is failed before NewWorkspace, per requirement.md L687 / L774. An empty or nil slice means "no whitelist" (all paths allowed) — matching the spec's "空配列の場合はホワイトリストを無効化（全パス許可）".

Prefix matching is byte-literal strings.HasPrefix; the config layer is responsible for any path normalisation (e.g. ~ expansion) before the slice reaches the dispatcher.

func WithAuditor 

func WithAuditor(a config.Auditor) Option

WithAuditor installs the audit-log sink. Every dispatch start and per-row failure records one event so requirement.md L29 invariant #2 "No silent execution" + L745 ("各ディスパッチで誰が何のタスクをいつ 何にディスパッチしたか・どの権限モードで起動したかを残す") are honoured. Defaults to config.NopAuditor so existing tests / CLI paths that have not yet wired audit.log keep building.

func WithBaseSkill 

func WithBaseSkill(s string) Option

WithBaseSkill sets the base execution skill content. Required.

func WithClaudeCommand 

func WithClaudeCommand(s string) Option

WithClaudeCommand sets the shell command cmux runs inside each new workspace. Required.

func WithClock 

func WithClock(now func() time.Time) Option

WithClock injects a deterministic clock for started_at writes. Defaults to time.Now in production.

func WithCmux 

func WithCmux(c cmux.Client) Option

WithCmux injects the cmux client. Required.

func WithLockKeys 

func WithLockKeys(m map[string]string) Option

WithLockKeys installs the [execution.lock_keys] regex map used to resolve notes.lock_hint. Optional; when absent, no AcquireLock call is ever issued and every row dispatches without contention.

func WithOnUnknownPermission 

func WithOnUnknownPermission(p string) Option

WithOnUnknownPermission selects the policy applied when the matcher denies a request. Accepts "escalate" / "fail" / "retry"; empty is treated as PermissionAsk.

func WithPermissionMatcher 

func WithPermissionMatcher(m PermissionMatcher) Option

WithPermissionMatcher installs the auto-accept allowlist resolver. Optional; when nil, HandlePermissionRequest returns PermissionAsk for every prompt (safe default — never silently allow).

func WithPermissionMode 

func WithPermissionMode(mode string) Option

WithPermissionMode declares the cfg.Execution.PermissionMode. New() uses it to enforce "non-bypass mode requires a PermissionMatcher".

func WithSourceSkill 

func WithSourceSkill(f SourceSkillFunc) Option

WithSourceSkill installs the source-specific skill resolver. Optional; when absent, source-specific guidance is omitted from every prompt.

func WithStore 

func WithStore(s Store) Option

WithStore injects the tasks-table repository. Required.

func WithTriageSkill 

func WithTriageSkill(s string) Option

WithTriageSkill injects the contents of skills/marunage-triage/SKILL.md (PR-34 distributes the embedded copy to ~/.claude/skills/ via `marunage setup --skills`). When set, BuildPrompt emits the triage section between the source-specific skill and the task block so the receiving Claude session loads OODA Orient guidance before reasoning about the task payload. Empty string disables the section entirely (back-compat for callers that have not wired the triage skill yet).

SECURITY: this string lands in the prompt as a TRUSTED section (BuildPrompt skips the fence-escape pass for it, just like Base / SourceSpecific). The caller MUST source it from the embedded skill file or its on-disk install — never from user input, task bodies, or any value derived from a discovery message — otherwise an attacker could splice forged fences into the prompt and bypass the protections fenceEscape gives the user-derived task fields.

func WithWorkspaceDirs 

func WithWorkspaceDirs(d WorkspaceDirs) Option

WithWorkspaceDirs installs the per-task control-directory provider. When set, the dispatcher mkdirs Dir(task.ID) before NewWorkspace and passes the same path into BuildPrompt so the dispatched Claude session writes its sentinel into the dir the PR-43 completion watcher is polling.

type PermissionDecision 

type PermissionDecision int

PermissionDecision is the dispatcher's verdict on one Claude tool permission request. The cmux/MCP shim that actually intercepts the prompt translates this into the protocol-level reply (allow / deny / re-prompt). This type lives in dispatch because the policy lives here too.

EXPORT NOTE: same as PermissionMatcher above — the only call site is internal until the cmux/MCP shim PR lands. 

const (
	// PermissionAllow: matcher accepted the (tool, args) pair.
	PermissionAllow PermissionDecision = iota
	// PermissionEscalate: matcher denied AND on_unknown_permission =
	// "escalate"; the row has been moved to waiting_human and a
	// dispatch.escalate audit event recorded.
	PermissionEscalate
	// PermissionFail: matcher denied AND on_unknown_permission =
	// "fail"; the row has been moved to failed and a dispatch.fail
	// audit event recorded.
	PermissionFail
	// PermissionAsk: the dispatcher will not decide. Either no matcher
	// is configured (safe default) or the policy is "retry" — the
	// caller is expected to surface the prompt back to a human or to
	// a re-prompt loop.
	PermissionAsk
)

func (PermissionDecision) String 

func (d PermissionDecision) String() string

String aids debug output and -v test failure messages.

type PermissionMatcher 

type PermissionMatcher interface {
	Allow(tool, args string) bool
}

PermissionMatcher abstracts the auto-accept allowlist resolver. The concrete implementation in internal/permission satisfies this; the interface lives here so test fakes do not need to construct a real permission.Matcher.

EXPORT NOTE: this interface is currently consumed only by the dispatcher itself. The eventual external consumer is the cmux/MCP shim PR (not yet on the plan) that intercepts Claude's tool prompts and forwards them to HandlePermissionRequest. If that PR slips indefinitely, consider re-narrowing this surface.

type PromptInputs 

type PromptInputs struct {
	// Base is the contents of skills/marunage-execute/SKILL.md (or the
	// caller's substitute). It always appears first so high-level
	// guardrails are loaded into context before anything else.
	Base string
	// SourceSpecific is the contents of skills/marunage-source-<name>/SKILL.md,
	// or empty when the source has no dedicated skill. Sandwiched between
	// Base and the task block so source-specific overrides can refine the
	// base instructions.
	SourceSpecific string
	// Triage is the contents of skills/marunage-triage/SKILL.md, or empty
	// when the dispatcher has not wired the triage skill (back-compat).
	// PR-72 inserts it between SourceSpecific and the task block so the
	// receiving Claude session loads the OODA Orient guidance before
	// reading the user-derived task payload it must classify.
	Triage string
	// Task is the row being dispatched. BuildPrompt reads ID / Source /
	// Title / Body and renders them into a labelled task block so the
	// receiving Claude session can quote them back deterministically.
	Task store.Task
	// WorkspaceDir is marunage's per-task control directory (typically
	// ~/.marunage/workspaces/<id>). When non-empty, BuildPrompt appends
	// a sentinel-write instruction telling Claude to publish completion
	// via `echo $? > .exit_code.tmp && mv .exit_code.tmp .exit_code`
	// inside this dir so the PR-43 completion watcher polling the same
	// path can detect exit. Empty disables the section entirely
	// (back-compat for callers that have not wired completion yet).
	WorkspaceDir string
}

PromptInputs is the ingredient list BuildPrompt assembles. Keeping it as a struct (rather than positional args) leaves room for future sections (system prompt overrides, reflection-on-resume, ...) without breaking call sites at the dispatcher.

type ReflectionStore 

type ReflectionStore interface {
	SetReflection(ctx context.Context, id int64, text string) error
}

ReflectionStore is the narrow read/write surface the Reflector needs against the tasks table. Production wires *store.TaskRepo; tests inject a fake. The Reflector takes the post-done store.Task as input from the completion watcher and never re-fetches, so SetReflection is the only method here — adding a Get just for "future flexibility" is YAGNI and only widens what fake implementations have to satisfy.

type Reflector 

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

Reflector is the PR-102 completion hook. It is invoked from the completion watcher (see internal/completion.WithDoneHook) immediately after a row flips to done. For the configured sample of completions it sends the marunage-reflect SKILL into the same cmux workspace and waits, asynchronously, for Claude to publish a `.reflection` sentinel in the workspace directory. When the file appears the trimmed body is persisted via store.SetReflection so a follow-up `marunage show` / Web UI render surfaces the answer next to the result_summary.

The hook fires in a fresh goroutine so the watcher's tick stays responsive. Wait blocks until every dispatched goroutine completes, which the daemon shutdown path uses to drain in-flight reflections before exiting.

func NewReflector 

func NewReflector(opts ...ReflectorOption) (*Reflector, error)

NewReflector validates required options and returns a Reflector ready to receive OnDone calls.

func (*Reflector) OnDone 

func (r *Reflector) OnDone(parent context.Context, task store.Task)

OnDone is the post-done hook. It samples; on accept it spawns a goroutine that sends the reflection prompt and waits for the `.reflection` sentinel to land in the per-task workspace dir. The caller must call Wait before exiting the process to drain in-flight reflections.

Empty task.WS is a no-op: there is no cmux session to send into.

func (*Reflector) Wait 

func (r *Reflector) Wait()

Wait blocks until every OnDone-spawned goroutine exits. Daemon shutdown should call Wait before closing the cmux connection so an in-flight reflection has a chance to finish (or hit its own timeout).

type ReflectorOption 

type ReflectorOption func(*reflectorBuild)

ReflectorOption mutates Reflector construction.

func WithReflectionAuditor 

func WithReflectionAuditor(a config.Auditor) ReflectorOption

WithReflectionAuditor installs the audit-log sink. Defaults to config.NopAuditor so callers that have not yet wired audit.log keep building.

func WithReflectionClock 

func WithReflectionClock(now func() time.Time) ReflectorOption

WithReflectionClock injects a deterministic clock for tests / future time-stamped audit fields. Defaults to time.Now.

func WithReflectionCmux 

func WithReflectionCmux(c cmux.Client) ReflectorOption

WithReflectionCmux injects the cmux client used to send the reflect prompt back into the same workspace. Required.

func WithReflectionPollInterval 

func WithReflectionPollInterval(d time.Duration) ReflectorOption

WithReflectionPollInterval overrides how often the goroutine stats the sentinel file. Defaults to 1s; tests squash to a few milliseconds.

func WithReflectionSampleRate 

func WithReflectionSampleRate(r float64) ReflectorOption

WithReflectionSampleRate installs a probability in [0,1]. Validated in NewReflector. Last-writer-wins against WithReflectionSampler: whichever option is supplied later in the NewReflector argument list determines the effective sampler.

func WithReflectionSampler 

func WithReflectionSampler(s Sampler) ReflectorOption

WithReflectionSampler injects an explicit Sampler — useful in tests (alwaysTrue / alwaysFalse) and for future strategies (deterministic hash of task id, time-of-day weighting, etc.). Last-writer-wins against WithReflectionSampleRate.

func WithReflectionSkill 

func WithReflectionSkill(s string) ReflectorOption

WithReflectionSkill installs the marunage-reflect SKILL.md body. The hook concatenates this body with a sentinel-write instruction before passing the result to cmux.Send.

func WithReflectionStore 

func WithReflectionStore(s ReflectionStore) ReflectorOption

WithReflectionStore injects the tasks-table repository. Required.

func WithReflectionTimeout 

func WithReflectionTimeout(d time.Duration) ReflectorOption

WithReflectionTimeout caps how long one OnDone goroutine waits for the .reflection sentinel before giving up. Defaults to 5m.

func WithReflectionWorkspaceDirs 

func WithReflectionWorkspaceDirs(d WorkspaceDirs) ReflectorOption

WithReflectionWorkspaceDirs injects the per-task control-directory resolver shared with the dispatcher and completion watcher. The goroutine reads the resulting `<dir>/.reflection` sentinel.

type RunOptions 

type RunOptions struct {
	// MaxParallel caps the number of pending rows dispatched in this Run.
	// Must be > 0 for any rows to dispatch — a zero/negative value is the
	// caller's signal of "do nothing". Ignored when ID > 0 (single-row
	// dispatch always processes exactly that row).
	MaxParallel int
	// ID restricts Run to the specified row, mirroring `marunage dispatch
	// <id>`. The row must currently be pending (post-lock skips and
	// failures still surface as errors / status writes the same way). A
	// zero value means "scan the pending queue".
	ID int64
}

RunOptions controls one Run invocation.

type Sampler 

type Sampler interface {
	Sample() bool
}

Sampler decides whether a given completion should trigger a reflection run. Returning false short-circuits OnDone before any cmux work fires. The default implementation is a rate-based random sampler; tests inject deterministic alwaysTrue / alwaysFalse stubs.

type SourceSkillFunc 

type SourceSkillFunc func(source string) string

SourceSkillFunc resolves the source-specific prompt skill (the contents of skills/marunage-source-<name>/SKILL.md). Returns "" for sources without a dedicated skill — BuildPrompt will then collapse the section cleanly.

type Store 

type Store interface {
	List(ctx context.Context, f store.ListFilter) ([]store.Task, error)
	Get(ctx context.Context, id int64) (store.Task, error)
	AcquireLock(ctx context.Context, id int64, lockKey string) error
	ReleaseLock(ctx context.Context, id int64) error
	ClaimWorkspace(ctx context.Context, id int64, ws string) (bool, error)
	SetWorkspace(ctx context.Context, id int64, ws string) error
	UpdateStatus(ctx context.Context, id int64, newStatus string) error
	SetStartedAt(ctx context.Context, id int64, t time.Time) error
	MarkFailedWithReason(ctx context.Context, id int64, reason string) error
	EscalateToHuman(ctx context.Context, id int64, reason string) error
}

Store is the narrow read/write surface the dispatcher needs against the tasks table. Keeping it as an interface (rather than the concrete *store.TaskRepo) is the test seam: production wires the real repo, tests can swap in a fake. The method set is intentionally a subset of *store.TaskRepo so the concrete type satisfies it implicitly.

type WorkspaceDirs 

type WorkspaceDirs interface {
	Dir(taskID int64) string
}

WorkspaceDirs resolves the per-task on-disk control directory marunage owns (typically ~/.marunage/workspaces/<id>). Used by the dispatcher to mkdir the dir before NewWorkspace and to embed the sentinel-write path in the prompt. Production wires the same concrete implementation the PR-43 completion watcher reads from, so the two layers cannot disagree on the path. The interface intentionally mirrors completion.WorkspaceDirs (same shape, same method) without importing it, to keep the dispatch -> completion dependency direction clean.

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.