Compare commits
3 Commits
v0.1.3
...
fe5074c3cf
| Author | SHA1 | Date | |
|---|---|---|---|
| fe5074c3cf | |||
| 8b30b9f889 | |||
| aab950f1c3 |
@@ -1,10 +1,11 @@
|
|||||||
# Gadfly — agentic adversarial PR reviewer (https://gitea.stevedudenhoeffer.com/steve/gadfly).
|
# Gadfly — agentic adversarial PR reviewer (https://gitea.stevedudenhoeffer.com/steve/gadfly).
|
||||||
#
|
#
|
||||||
# Runs the published Gadfly image (pinned to :v1) as a specialist swarm and posts
|
# Runs the published Gadfly image (pinned to an immutable :sha- tag — act_runner
|
||||||
|
# caches :latest, and this build is what carries foreman provider-type support)
|
||||||
|
# as a specialist swarm and posts
|
||||||
# ONE consolidated review comment as gitea-actions. Advisory only — never blocks a
|
# ONE consolidated review comment as gitea-actions. Advisory only — never blocks a
|
||||||
# merge. This reviews executus PRs (same setup as mort: m1/m5 locals + 2 cloud,
|
# merge. This reviews executus PRs (same setup as mort: m1/m5 foreman locals + 2
|
||||||
# 3-lens suite). Gadfly is a simple system — treat its findings as advisory and
|
# cloud, 3-lens suite). Gadfly is a simple system — findings are advisory; double-check.
|
||||||
# double-check before acting.
|
|
||||||
|
|
||||||
name: Adversarial Review (Gadfly)
|
name: Adversarial Review (Gadfly)
|
||||||
|
|
||||||
@@ -44,20 +45,31 @@ jobs:
|
|||||||
# every PR with the 3-lens suite — the slow local lanes dominate wall time.
|
# every PR with the 3-lens suite — the slow local lanes dominate wall time.
|
||||||
timeout-minutes: 90
|
timeout-minutes: 90
|
||||||
steps:
|
steps:
|
||||||
- uses: docker://gitea.stevedudenhoeffer.com/steve/gadfly:v1
|
- uses: docker://gitea.stevedudenhoeffer.com/steve/gadfly:sha-6e3a83c
|
||||||
env:
|
env:
|
||||||
GITEA_API: ${{ github.server_url }}/api/v1/repos/${{ github.repository }}
|
GITEA_API: ${{ github.server_url }}/api/v1/repos/${{ github.repository }}
|
||||||
GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
|
GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
|
||||||
OLLAMA_CLOUD_API_KEY: ${{ secrets.OLLAMA_CLOUD_API_KEY }}
|
OLLAMA_CLOUD_API_KEY: ${{ secrets.OLLAMA_CLOUD_API_KEY }}
|
||||||
# Local Ollama boxes (each its own lane, cap 1). NOTE: both Macs must be
|
# Local Macs, reached through their foreman queues (native Ollama on the
|
||||||
# awake/reachable for their reviews to run; if a box is offline, that
|
# wire). Gadfly's GADFLY_ENDPOINT_* form with the "foreman" provider
|
||||||
# model's comment shows an error and the others still post.
|
# type: GADFLY_ENDPOINT_M1 registers provider "m1", _M5 registers "m5",
|
||||||
GADFLY_ENDPOINT_M1PRO: "ollama|http://192.168.0.175:11434"
|
# each building a foreman-preset Ollama client at the given URL. Values
|
||||||
GADFLY_ENDPOINT_M5MAX: "ollama|http://192.168.0.173:11434"
|
# (host + token) live in gitea secrets, each of the form:
|
||||||
|
# foreman|https://<foreman-host>|<token>
|
||||||
|
# (converted from the komodo LLM_* DSNs foreman://<token>@<host>).
|
||||||
|
# REQUIRES a Gadfly image built with foreman provider-type support
|
||||||
|
# (the GADFLY_ENDPOINT "foreman|..." type); on an older image the m1/m5
|
||||||
|
# lanes error with "unknown provider foreman". The HTTPS-only LLM_*
|
||||||
|
# foreman:// DSN is the alternative that needs no image rebuild.
|
||||||
|
# NOTE: the Mac behind each foreman must still be awake/reachable; if a
|
||||||
|
# box is offline, that model's comment shows an error and the others
|
||||||
|
# still post. (Gitea secrets aren't auto-exposed — map each explicitly.)
|
||||||
|
GADFLY_ENDPOINT_M1: ${{ secrets.GADFLY_ENDPOINT_M1 }}
|
||||||
|
GADFLY_ENDPOINT_M5: ${{ secrets.GADFLY_ENDPOINT_M5 }}
|
||||||
# 2 cloud (parallel) + M1 Pro + M5 Max — one consolidated comment each.
|
# 2 cloud (parallel) + M1 Pro + M5 Max — one consolidated comment each.
|
||||||
GADFLY_MODELS: "minimax-m3:cloud,deepseek-v4-flash:cloud,m1pro/qwen3:14b,m5max/qwen3.6:35b-mlx"
|
GADFLY_MODELS: "minimax-m3:cloud,deepseek-v4-flash:cloud,m1/qwen3:14b,m5/qwen3.6:35b-mlx"
|
||||||
# cloud runs 2 at once; each Mac one at a time; all three lanes parallel.
|
# cloud runs 2 at once; each Mac one at a time; all three lanes parallel.
|
||||||
GADFLY_PROVIDER_CONCURRENCY: "ollama-cloud=2,m1pro=1,m5max=1"
|
GADFLY_PROVIDER_CONCURRENCY: "ollama-cloud=2,m1=1,m5=1"
|
||||||
# Default => the 3-lens suite (security, correctness, error-handling).
|
# Default => the 3-lens suite (security, correctness, error-handling).
|
||||||
# Set the repo var GADFLY_SPECIALISTS to override (csv / "all" / "auto").
|
# Set the repo var GADFLY_SPECIALISTS to override (csv / "all" / "auto").
|
||||||
GADFLY_SPECIALISTS: ${{ vars.GADFLY_SPECIALISTS || 'security,correctness,error-handling' }}
|
GADFLY_SPECIALISTS: ${{ vars.GADFLY_SPECIALISTS || 'security,correctness,error-handling' }}
|
||||||
|
|||||||
@@ -43,8 +43,11 @@ CORE (majordomo + stdlib):
|
|||||||
fanout/ programmatic N×M swarm [P0 ✓]
|
fanout/ programmatic N×M swarm [P0 ✓]
|
||||||
deliver/ output egress seam (+ Discard/Stdout) [P0 ✓]
|
deliver/ output egress seam (+ Discard/Stdout) [P0 ✓]
|
||||||
identity/ caller identity seams [P0 ✓]
|
identity/ caller identity seams [P0 ✓]
|
||||||
run/ progress bridge now; the executor kernel + [P0 partial]
|
run/ run-loop mechanics + RunnableAgent DTO + [P2 wip]
|
||||||
nil-safe Ports + RunnableAgent later [P2]
|
nil-safe run.Ports (Audit/Budget/Critic/
|
||||||
|
Checkpointer/PaletteSource) defined; the
|
||||||
|
agentexec+skillexec -> run.Executor MERGE
|
||||||
|
(consuming Ports) is the remaining P2 work [P2]
|
||||||
dispatchguard/ loop/depth/fan-out caps [P0 ✓]
|
dispatchguard/ loop/depth/fan-out caps [P0 ✓]
|
||||||
pendingattach/ attachment dedupe [P0 ✓]
|
pendingattach/ attachment dedupe [P0 ✓]
|
||||||
tool/ registry + 3-stage permissions + ssrf [P1 ✓]
|
tool/ registry + 3-stage permissions + ssrf [P1 ✓]
|
||||||
|
|||||||
@@ -0,0 +1,76 @@
|
|||||||
|
package run
|
||||||
|
|
||||||
|
import "time"
|
||||||
|
|
||||||
|
// RunnableAgent is the kernel's view of "a thing to run": an identity, a model
|
||||||
|
// tier, a system prompt, execution caps, and a tool palette. It is a plain DTO
|
||||||
|
// on purpose — the run kernel never imports a noun battery. The persona Agent
|
||||||
|
// and the saved Skill each LOWER themselves into a RunnableAgent (a ToRunnable
|
||||||
|
// method on the battery side), and the kernel runs the DTO. This is the
|
||||||
|
// inversion of mort's agentexec.Executor.Run(*agents.Agent): the executor no
|
||||||
|
// longer depends on the persona struct, only on this shape.
|
||||||
|
//
|
||||||
|
// A light host can build a RunnableAgent inline (model tier + prompt + a few
|
||||||
|
// tool names) for a one-shot bounded run, with no persona or skill battery at
|
||||||
|
// all — that is exactly gadfly's swarm task.
|
||||||
|
type RunnableAgent struct {
|
||||||
|
// ID is a stable identifier for the run subject (an agent/skill UUID, or
|
||||||
|
// any host-chosen id). Used for audit attribution and dispatch-guard
|
||||||
|
// genealogy. Empty is allowed for anonymous one-shot runs.
|
||||||
|
ID string
|
||||||
|
|
||||||
|
// Name is a human label (audit/logs/delivery). Empty is allowed.
|
||||||
|
Name string
|
||||||
|
|
||||||
|
// SystemPrompt is the agent's base system prompt (before per-run
|
||||||
|
// personalization, which a host layers via Ports).
|
||||||
|
SystemPrompt string
|
||||||
|
|
||||||
|
// ModelTier is a tier alias or concrete spec resolved through
|
||||||
|
// model.ParseModelForContext. Empty resolves to the host's default tier.
|
||||||
|
ModelTier string
|
||||||
|
|
||||||
|
// MaxIterations caps the agent loop's tool-dispatch steps. 0 = kernel
|
||||||
|
// default. MaxRuntime caps wall-clock for the whole run (the kernel starts
|
||||||
|
// this clock AFTER any lane dequeue, not at submission). 0 = kernel
|
||||||
|
// default.
|
||||||
|
MaxIterations int
|
||||||
|
MaxRuntime time.Duration
|
||||||
|
|
||||||
|
// LowLevelTools are tool-registry names the run may call directly.
|
||||||
|
// SkillPalette / SubAgentPalette name saved skills / sub-agents exposed as
|
||||||
|
// skill__<name> / agent__<name> delegation tools, resolved through
|
||||||
|
// Ports.Palette (nil Palette => those entries are inert).
|
||||||
|
LowLevelTools []string
|
||||||
|
SkillPalette []string
|
||||||
|
SubAgentPalette []string
|
||||||
|
|
||||||
|
// Phases optionally model a multi-step pipeline (each phase its own prompt
|
||||||
|
// + tier + tools). An empty slice is a single-phase run — the common case.
|
||||||
|
Phases []Phase
|
||||||
|
|
||||||
|
// Critic configures the optional two-tier run-critic (Ports.Critic). The
|
||||||
|
// zero value (disabled) is the light-host default.
|
||||||
|
Critic CriticConfig
|
||||||
|
}
|
||||||
|
|
||||||
|
// Phase is one step of a multi-step run: its own system prompt, model tier,
|
||||||
|
// iteration cap, and tool subset. Optional phases may be skipped by the
|
||||||
|
// pipeline when their precondition isn't met.
|
||||||
|
type Phase struct {
|
||||||
|
Name string
|
||||||
|
SystemPrompt string
|
||||||
|
ModelTier string
|
||||||
|
MaxIterations int
|
||||||
|
Tools []string
|
||||||
|
Optional bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// CriticConfig configures the optional run-critic. Enabled gates whether a
|
||||||
|
// critic monitor is started at all; BackstopMultiplier sets the hard-kill
|
||||||
|
// deadline as a multiple of the soft trigger (MaxRuntime). A non-positive
|
||||||
|
// multiplier uses the kernel default.
|
||||||
|
type CriticConfig struct {
|
||||||
|
Enabled bool
|
||||||
|
BackstopMultiplier float64
|
||||||
|
}
|
||||||
+168
@@ -0,0 +1,168 @@
|
|||||||
|
package run
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
|
||||||
|
|
||||||
|
"gitea.stevedudenhoeffer.com/steve/executus/deliver"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Ports are the host seams the run executor consumes. Every field is nil-safe:
|
||||||
|
// a light host passes the zero Ports and gets a bounded, in-memory run with no
|
||||||
|
// persistence, audit, budget, critic, delegation, or delivery — which is
|
||||||
|
// exactly a gadfly swarm task. A heavy host (mort) wires each one to a battery.
|
||||||
|
//
|
||||||
|
// This struct IS the inversion: in mort, agentexec imports agents /
|
||||||
|
// agentcritic / skillaudit and skillexec imports skills / paste directly; here
|
||||||
|
// the kernel depends only on these interfaces, and the batteries implement
|
||||||
|
// them. The mort_*_adapters.go wall becomes the set of impls.
|
||||||
|
type Ports struct {
|
||||||
|
// Audit records the run trace (start, per-step/per-tool events, final
|
||||||
|
// stats). nil = no audit.
|
||||||
|
Audit Audit
|
||||||
|
// Budget gates and meters per-caller resource use. nil = unbounded.
|
||||||
|
Budget Budget
|
||||||
|
// Critic optionally monitors a long run for hangs/runaways. nil = none.
|
||||||
|
Critic Critic
|
||||||
|
// Checkpointer persists resumable progress for durable recovery. nil = no
|
||||||
|
// checkpointing (a run interrupted by shutdown is simply lost).
|
||||||
|
Checkpointer Checkpointer
|
||||||
|
// Palette resolves SkillPalette / SubAgentPalette entries into delegation
|
||||||
|
// tools (skill__<name> / agent__<name>). nil = those entries are inert.
|
||||||
|
Palette PaletteSource
|
||||||
|
// Delivery is where the run's output + artifacts go. nil = the caller
|
||||||
|
// reads the Result in-process (the light-host default).
|
||||||
|
Delivery deliver.Delivery
|
||||||
|
}
|
||||||
|
|
||||||
|
// RunInfo describes a run at start time — the attribution a recorder/critic
|
||||||
|
// needs. Host-neutral rename of mort's SkillRun start fields.
|
||||||
|
type RunInfo struct {
|
||||||
|
RunID string
|
||||||
|
SubjectID string // the agent/skill id being run (audit "skill_id")
|
||||||
|
Name string
|
||||||
|
CallerID string
|
||||||
|
ChannelID string
|
||||||
|
ParentRunID string
|
||||||
|
Inputs map[string]any
|
||||||
|
StartedAt time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// RunStats is the terminal roll-up a recorder's Close writes. Mirrors mort's
|
||||||
|
// skillaudit/skillexec RunStats.
|
||||||
|
type RunStats struct {
|
||||||
|
Status string // ok | error | timeout | budget_exceeded | cancelled | dry_run
|
||||||
|
Output string
|
||||||
|
Error string
|
||||||
|
ToolCalls int
|
||||||
|
RuntimeSeconds float64
|
||||||
|
InputTokens int64
|
||||||
|
OutputTokens int64
|
||||||
|
ThinkingTokens int64
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Audit ---
|
||||||
|
|
||||||
|
// Audit begins recording a run. StartRun returns a per-run RunRecorder (or nil
|
||||||
|
// to skip recording this run). The audit battery wires its Storage behind this.
|
||||||
|
type Audit interface {
|
||||||
|
StartRun(ctx context.Context, info RunInfo) RunRecorder
|
||||||
|
}
|
||||||
|
|
||||||
|
// RunRecorder records the events of one in-flight run and its final stats. It
|
||||||
|
// satisfies RunTally so the kernel can surface live token/tool counts to the
|
||||||
|
// self-status tool. Mirrors mort's skillaudit.Writer.
|
||||||
|
type RunRecorder interface {
|
||||||
|
RunTally
|
||||||
|
// OnStep records one completed agent-loop iteration's model response.
|
||||||
|
OnStep(iter int, resp *llm.Response)
|
||||||
|
// OnTool records one executed tool call + its result.
|
||||||
|
OnTool(call llm.ToolCall, result string)
|
||||||
|
// LogEvent / LogError append structured events to the run log.
|
||||||
|
LogEvent(eventType string, payload map[string]any)
|
||||||
|
LogError(msg string)
|
||||||
|
// Close writes the terminal roll-up. Detaches from the caller's context
|
||||||
|
// internally so a cancelled run still records.
|
||||||
|
Close(ctx context.Context, stats RunStats)
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Budget ---
|
||||||
|
|
||||||
|
// Budget gates and meters per-caller resource use. Mirrors mort's
|
||||||
|
// skillexec.BudgetTracker.
|
||||||
|
type Budget interface {
|
||||||
|
// Check reports whether the caller has remaining budget (nil = allowed).
|
||||||
|
Check(ctx context.Context, callerID string) error
|
||||||
|
// Commit records that the caller spent runtimeSeconds on this run.
|
||||||
|
Commit(ctx context.Context, callerID string, runtimeSeconds float64)
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Critic ---
|
||||||
|
|
||||||
|
// Critic optionally monitors a long-running run (the two-tier soft/hard
|
||||||
|
// timeout). Monitor returns a handle the executor feeds progress into and
|
||||||
|
// queries for steer/deadline decisions; a nil handle means "not monitored".
|
||||||
|
//
|
||||||
|
// The exact wiring (how the handle's Steer/Deadline bind into majordomo's
|
||||||
|
// agent.WithSteer / agent.WithMaxStepsFunc / run-context cancellation) is
|
||||||
|
// finalized in the executor; this is the seam the agentcritic battery adapts.
|
||||||
|
type Critic interface {
|
||||||
|
Monitor(ctx context.Context, info RunInfo, softTimeout time.Duration) CriticHandle
|
||||||
|
}
|
||||||
|
|
||||||
|
// CriticHandle is the executor's live link to a run's critic.
|
||||||
|
type CriticHandle interface {
|
||||||
|
// RecordStep / RecordToolStart keep the critic's activity clock fresh so a
|
||||||
|
// healthy-but-slow run is not mistaken for a hang.
|
||||||
|
RecordStep(iter int)
|
||||||
|
RecordToolStart(name, args string)
|
||||||
|
// Steer returns any messages the critic wants injected into the loop (a
|
||||||
|
// nudge), drained before each step — matches majordomo agent.WithSteer.
|
||||||
|
Steer() []llm.Message
|
||||||
|
// Deadline returns the current hard-kill deadline (the critic may extend
|
||||||
|
// it); the executor binds the run context to it. Zero = no hard deadline.
|
||||||
|
Deadline() time.Time
|
||||||
|
// Stop ends monitoring when the run finishes.
|
||||||
|
Stop()
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Checkpointer ---
|
||||||
|
|
||||||
|
// Checkpointer persists a run's resumable progress for durable recovery.
|
||||||
|
// Mirrors mort's agentexec.RunCheckpointer.
|
||||||
|
type Checkpointer interface {
|
||||||
|
// Save persists the run's current resumable progress (throttled).
|
||||||
|
Save(ctx context.Context, st RunCheckpointState) error
|
||||||
|
// Complete clears the checkpoint on success.
|
||||||
|
Complete(ctx context.Context) error
|
||||||
|
// Fail clears the checkpoint on terminal failure. A run interrupted by
|
||||||
|
// shutdown is left untouched so boot recovery picks it up.
|
||||||
|
Fail(ctx context.Context, err error) error
|
||||||
|
}
|
||||||
|
|
||||||
|
// RunCheckpointState is the resumable snapshot a Checkpointer persists. Kept
|
||||||
|
// minimal here; the executor extends what it records during the merge.
|
||||||
|
type RunCheckpointState struct {
|
||||||
|
Messages []llm.Message
|
||||||
|
Iteration int
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- PaletteSource ---
|
||||||
|
|
||||||
|
// PaletteSource resolves a RunnableAgent's SkillPalette / SubAgentPalette names
|
||||||
|
// into delegation tools and invokes them. Mirrors mort's
|
||||||
|
// SkillInvokerForPalette + AgentInvokerForPalette. nil Palette => palette
|
||||||
|
// entries are inert ("not configured" at first call).
|
||||||
|
type PaletteSource interface {
|
||||||
|
ResolveSkill(ctx context.Context, callerID, name string) (skillID string, err error)
|
||||||
|
InvokeSkill(ctx context.Context, callerID, channelID, name string,
|
||||||
|
inputs map[string]any, parentRunID string) (output, runID, status string, err error)
|
||||||
|
|
||||||
|
ResolveAgent(ctx context.Context, callerID, name string) (agentID string, err error)
|
||||||
|
InvokeAgent(ctx context.Context, callerID, channelID, name string,
|
||||||
|
prompt, parentRunID, modelTierOverride, promptPrepend string,
|
||||||
|
toolsSubset []string,
|
||||||
|
onEvent func(ctx context.Context, event, emoji string)) (output, runID, status string, err error)
|
||||||
|
}
|
||||||
@@ -0,0 +1,157 @@
|
|||||||
|
// Package run is executus's run kernel: the shared run-loop mechanics around
|
||||||
|
// majordomo's agent loop, plus the host seams (run.Ports / RunnableAgent) that
|
||||||
|
// let one executor serve every surface — a light host's bounded one-shot run,
|
||||||
|
// a heavy host's persona agent or saved skill — without the kernel importing a
|
||||||
|
// battery.
|
||||||
|
//
|
||||||
|
// This file holds the genuinely-identical scaffolding both run shapes need:
|
||||||
|
// context cancellation merging, the detached-cleanup timeout, the per-run
|
||||||
|
// progress accessor the self-status tool reads, the legacy `submit`
|
||||||
|
// compatibility tool (submit.go), the ancestor progress bridge (progress.go),
|
||||||
|
// and the run-finalizer machinery — one source of truth.
|
||||||
|
//
|
||||||
|
// The kernel depends only on majordomo + executus/tool + the run.Ports
|
||||||
|
// interfaces; persistence, audit, the persona/skill nouns, and the critic are
|
||||||
|
// host-supplied via Ports (see ports.go) so importing the kernel never drags in
|
||||||
|
// a store or a battery.
|
||||||
|
package run
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"log/slog"
|
||||||
|
"sync/atomic"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.stevedudenhoeffer.com/steve/executus/tool"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ErrShutdown is the cancellation cause set on mort's base lifecycle context
|
||||||
|
// when the process is shutting down (SIGTERM after the drain window). The
|
||||||
|
// agent executor uses it to distinguish a run interrupted by shutdown (which
|
||||||
|
// should be left durable-recoverable) from a run that errored or hit its own
|
||||||
|
// deadline (terminal).
|
||||||
|
var ErrShutdown = errors.New("mort: shutting down")
|
||||||
|
|
||||||
|
// CleanupContextTimeout caps how long a run's post-completion cleanup ops
|
||||||
|
// (budget commit, audit Close, attachment bookkeeping) may wait on
|
||||||
|
// storage after detaching from the caller's — possibly already
|
||||||
|
// cancelled — context. 10s is generous for a single-row UPDATE against
|
||||||
|
// MySQL; longer suggests a hung connection the run goroutine shouldn't
|
||||||
|
// keep waiting on. Both executors derive their cleanup contexts as
|
||||||
|
// context.WithTimeout(context.WithoutCancel(ctx), CleanupContextTimeout).
|
||||||
|
const CleanupContextTimeout = 10 * time.Second
|
||||||
|
|
||||||
|
// Reserved state-react lifecycle event keys, shared so both nouns surface
|
||||||
|
// the same UX shape. Namespaced with double-underscores to make accidental
|
||||||
|
// collision with a tool name near-impossible.
|
||||||
|
const (
|
||||||
|
StateReactStart = "__start__"
|
||||||
|
StateReactEnd = "__end__"
|
||||||
|
StateReactError = "__error__"
|
||||||
|
StateReactBudgetExceeded = "__budget_exceeded__"
|
||||||
|
)
|
||||||
|
|
||||||
|
// MergeCancellation returns a context cancelled when EITHER input is
|
||||||
|
// cancelled, propagating the cancellation Cause from whichever fired. Used
|
||||||
|
// by the lane preemption path (the lane's per-job ctx.Cause flows into the
|
||||||
|
// run context) and by the runtime-detach path (process shutdown still
|
||||||
|
// reaches a run whose deadline was reset after a lane wait). Always call
|
||||||
|
// the returned cancel to release the watcher goroutine; it is also invoked
|
||||||
|
// once when either input fires.
|
||||||
|
func MergeCancellation(parent, secondary context.Context) (context.Context, context.CancelFunc) {
|
||||||
|
merged, cancel := context.WithCancelCause(parent)
|
||||||
|
go func() {
|
||||||
|
select {
|
||||||
|
case <-merged.Done():
|
||||||
|
return
|
||||||
|
case <-secondary.Done():
|
||||||
|
cancel(context.Cause(secondary))
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
return merged, func() { cancel(nil) }
|
||||||
|
}
|
||||||
|
|
||||||
|
// RunFinalizer is invoked at run finish so per-run tool state (open HTTP
|
||||||
|
// streams, per-run code_exec counters, per-run search budgets) is released
|
||||||
|
// and the process-lifetime maps keyed by run id don't grow unbounded.
|
||||||
|
// Both executors fire their registered finalizers via FireFinalizers.
|
||||||
|
type RunFinalizer interface {
|
||||||
|
FinalizeRun(runID string)
|
||||||
|
}
|
||||||
|
|
||||||
|
// FireFinalizers runs every finalizer for runID, isolating each behind a
|
||||||
|
// panic-recover so one buggy finalizer can't take down the run goroutine
|
||||||
|
// or skip the others. Safe to call with a nil/empty slice.
|
||||||
|
func FireFinalizers(fs []RunFinalizer, runID string) {
|
||||||
|
for _, f := range fs {
|
||||||
|
if f == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
func() {
|
||||||
|
defer func() {
|
||||||
|
if r := recover(); r != nil {
|
||||||
|
slog.Error("runengine: run finalizer panicked",
|
||||||
|
"run_id", runID, "panic", r)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
f.FinalizeRun(runID)
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// RunTally is the narrow live-progress source the RunStateAccessor reads —
|
||||||
|
// the running token and tool-call counts for the in-flight run. The audit
|
||||||
|
// battery's writer satisfies it; this interface is how the run kernel reads
|
||||||
|
// live tallies without importing the audit package (the inversion of mort's
|
||||||
|
// direct *skillaudit.Writer dependency).
|
||||||
|
type RunTally interface {
|
||||||
|
// TokenStats returns the running input, output, and thinking token totals.
|
||||||
|
TokenStats() (in, out, thinking int64)
|
||||||
|
// ToolCallsCount returns the number of tool calls executed so far.
|
||||||
|
ToolCallsCount() int
|
||||||
|
}
|
||||||
|
|
||||||
|
// RunStateAccessor is the per-run live-progress accessor the executor
|
||||||
|
// stamps on Invocation.RunState before building the toolbox, so the
|
||||||
|
// self-status tool can report iteration / tool-calls / tokens / elapsed for
|
||||||
|
// the in-flight run. Construct with NewRunStateAccessor; the executor's step
|
||||||
|
// observer calls SetIteration each loop.
|
||||||
|
type RunStateAccessor struct {
|
||||||
|
tally RunTally
|
||||||
|
iter atomic.Int32
|
||||||
|
maxIter int
|
||||||
|
maxCalls int
|
||||||
|
startedAt time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewRunStateAccessor builds the accessor. writer supplies the live token
|
||||||
|
// + tool-call tallies; maxIter / maxCalls are the reported caps (0 =
|
||||||
|
// uncapped); startedAt anchors the elapsed clock.
|
||||||
|
func NewRunStateAccessor(tally RunTally, maxIter, maxCalls int, startedAt time.Time) *RunStateAccessor {
|
||||||
|
return &RunStateAccessor{
|
||||||
|
tally: tally,
|
||||||
|
maxIter: maxIter,
|
||||||
|
maxCalls: maxCalls,
|
||||||
|
startedAt: startedAt,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetIteration records the current agent-loop iteration (called from the
|
||||||
|
// executor's step observer).
|
||||||
|
func (a *RunStateAccessor) SetIteration(iter int) { a.iter.Store(int32(iter)) }
|
||||||
|
|
||||||
|
// RunState satisfies tool.RunStateAccessor.
|
||||||
|
func (a *RunStateAccessor) RunState() tool.RunState {
|
||||||
|
in, out, think := a.tally.TokenStats()
|
||||||
|
return tool.RunState{
|
||||||
|
Iteration: int(a.iter.Load()),
|
||||||
|
MaxIterations: a.maxIter,
|
||||||
|
ToolCalls: a.tally.ToolCallsCount(),
|
||||||
|
MaxToolCalls: a.maxCalls,
|
||||||
|
InputTokens: in,
|
||||||
|
OutputTokens: out,
|
||||||
|
ThinkingTokens: think,
|
||||||
|
ElapsedSeconds: int(time.Since(a.startedAt).Seconds()),
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,84 @@
|
|||||||
|
package run
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
|
||||||
|
llm "gitea.stevedudenhoeffer.com/steve/majordomo/llm"
|
||||||
|
)
|
||||||
|
|
||||||
|
// SubmitCapture records the output a run's `submit` tool received.
|
||||||
|
//
|
||||||
|
// Why this exists: legacy agentkit injected a synthetic `submit` tool and
|
||||||
|
// ended the loop when it fired; years of mort system prompts (agent
|
||||||
|
// YAMLs, skill manifests, the executors' platform headers) teach the
|
||||||
|
// model to "call submit with your final answer". majordomo's agent loop
|
||||||
|
// has no submit concept — it ends when the model replies WITHOUT tool
|
||||||
|
// calls. Dropping submit cold would make every prompt-trained model
|
||||||
|
// burn turns on "unknown tool \"submit\"" errors.
|
||||||
|
//
|
||||||
|
// The compatibility shape: the executors add NewSubmitTool's tool to
|
||||||
|
// every run's toolset (unless the palette already defines a `submit`).
|
||||||
|
// The handler records the FIRST submitted answer and tells the model
|
||||||
|
// the answer was accepted so its next turn is a bare reply (which ends
|
||||||
|
// the loop naturally). After the run, the executor consults
|
||||||
|
// Output(loopOutput, runErr): a captured submission wins over an empty
|
||||||
|
// or budget-exhausted ending, so a model that submits on its final
|
||||||
|
// allowed step still produces its answer instead of ErrMaxSteps.
|
||||||
|
type SubmitCapture struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
output string
|
||||||
|
called bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// Record stores the first submitted answer; later calls are ignored
|
||||||
|
// (matching legacy agentkit's "multiple calls keep the first" contract).
|
||||||
|
func (c *SubmitCapture) Record(output string) {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
if c.called {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
c.called = true
|
||||||
|
c.output = output
|
||||||
|
}
|
||||||
|
|
||||||
|
// Submitted returns the captured answer and whether submit fired.
|
||||||
|
func (c *SubmitCapture) Submitted() (string, bool) {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
return c.output, c.called
|
||||||
|
}
|
||||||
|
|
||||||
|
// Output resolves the run's final output: the submitted answer when the
|
||||||
|
// model called submit (parity with legacy agentkit, where submit's argument
|
||||||
|
// WAS the run output), otherwise the loop's own final text. resolvedErr
|
||||||
|
// is nil when a submission exists — a run that submitted its answer and
|
||||||
|
// then ran out of steps (or timed out composing the courtesy
|
||||||
|
// confirmation turn) is a SUCCESS, not an error.
|
||||||
|
func (c *SubmitCapture) Output(loopOutput string, runErr error) (output string, resolvedErr error) {
|
||||||
|
if out, ok := c.Submitted(); ok {
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
return loopOutput, runErr
|
||||||
|
}
|
||||||
|
|
||||||
|
// submitArgs mirrors legacy agentkit's synthetic submit tool schema so
|
||||||
|
// models prompted under the old contract emit compatible calls.
|
||||||
|
type submitArgs struct {
|
||||||
|
Output string `json:"output" description:"The final answer, summary, or output for this task."`
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewSubmitTool builds the compatibility `submit` tool bound to the
|
||||||
|
// given capture. Both executors (skill + agent) install one per run.
|
||||||
|
func NewSubmitTool(capture *SubmitCapture) llm.Tool {
|
||||||
|
return llm.DefineTool[submitArgs](
|
||||||
|
"submit",
|
||||||
|
"Submit your final answer or output to end this task. Call exactly once when you are done.",
|
||||||
|
func(_ context.Context, args submitArgs) (any, error) {
|
||||||
|
capture.Record(strings.TrimSpace(args.Output))
|
||||||
|
return "Final answer recorded. Do not call any more tools; reply now with a brief closing message.", nil
|
||||||
|
},
|
||||||
|
)
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user