10 Commits

Author SHA1 Message Date
steve 89f3334512 P3: meta + primitive tool group (think/now/cite + classify/extract/summarize)
Grow executus/tools into a real generic tool library:

- Register(reg): the always-available, zero-config tools — think, now (UTC
  unless a CurrentTimeProvider is wired), cite (inert unless a CitationStorage
  is wired). All nil-safe; a light host calls Register and is useful.
- RegisterMeta(reg, MetaDeps): the LLM-backed meta tools — classify,
  extract_entities, summarize — over the llmmeta helper. Budget defaults to the
  shipped in-memory per-run cap; Files optional; caps default.
- Seams moved (interface/type-only, no host coupling): research_providers.go
  (CurrentTimeProvider/CitationStorage/SearchBudget/PageExtractor/PDFFetcher/…)
  and file_storage.go (FileStorage + FileDomainMeta). Plus the in-memory budget
  default (research_defaults.go) and scope_validate.go.

calculate deferred (drags github.com/Krognol/go-wolfram + a module-path replace
— not worth it in the lean core for one tool). Core go.sum still free of
gorm/redis/discordgo/sqlite/wolfram.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 22:02:54 -04:00
steve fae1dcecad P3 (kickoff): generic tools/ library + end-to-end tool-using-agent test
Stand up executus/tools — the generic, host-agnostic tool library — and prove
the full pattern end to end:

- tools/tools.go: Register(reg) adds the always-available zero-dependency tools
  (currently `think`). A light host calls it and is immediately useful; backed
  tools (web/store/meta groups) will register via grouped registrars with
  nil-safe Deps as they land.
- tools/think.go: the `think` tool moved from mort (imports only executus/tool).
- tools/integration_test.go: end-to-end proof that the executor runs an agent
  which CALLS a registered tool — the fake model emits a `think` tool call, the
  executor dispatches it through the registry, the model finalises, and the step
  instrumentation captures the `think` step. Exercises the full tool-dispatch
  loop through run.Executor.

Stacked on phase-2-run-kernel (P3 needs run.Executor). Remaining P3: the
meta/web/net/store/compose groups + their Deps + default backends (splitting
mort's default.go grab-bag).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 22:02:54 -04:00
steve b35514dfaa ci(gadfly): cloud-only fleet (3 models, drop local Macs)
executus CI / test (push) Successful in 57s
Measured on the P2 review: the local Macs (m1/m5) took 26–29 min with lens
timeouts and found ZERO real bugs, while the two cloud models found every
genuine finding in 6–12 min. Drop the Macs; add glm-5.2:cloud as a third
cloud reviewer. Net: faster (~29→~12 min) and higher signal.

Models: minimax-m3:cloud, deepseek-v4-flash:cloud, glm-5.2:cloud
(ollama-cloud=3 concurrency). timeout-minutes 90→30.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 02:02:21 +00:00
steve 7b3da87c08 fix: address verified gadfly P2 findings (9 real of 18)
Independently verified all 18 gadfly findings against the code (18-agent
fan-out). Fixed the 9 real ones; the other 9 were false-positive /
hallucinated / valid-tradeoff (no change).

High:
- F1 nil model: a Models resolver returning (ctx,nil,nil) flowed into the
  agent loop and nil-panicked. Now a clean error (Run never panics). +test.
- F9 compactor data-leak: renderTranscript sent tool-call args verbatim to
  the summarizer (a possibly-different provider/tier); secret-bearing tool
  args (mcp_call/email_send/http_*/webhook_*) are now redacted, with a doc
  note that result bodies still flow (summary needs them).

Medium/minor:
- F2 compactor error path returned the folded slice, not the original msgs
  (contradicting the documented non-fatal contract) -> return msgs.
- F3 RunStats.Status only ok/error; now timeout (DeadlineExceeded) /
  cancelled (Canceled) via statusFor. +test.
- F4 step-zip emitted empty-name "ghost" steps when results>calls; now pairs
  min(calls,results) only.
- F5 SetIteration was never called -> RunState.Iteration always 0; the step
  observer now updates it each loop.
- F6 matchPending fallback was LIFO; now FIFO (matches the per-key queue).
- F7 estimateTokens had no default arm (future Part kinds counted as 0);
  unknown parts now counted conservatively.
- F8 cloud_sync silently truncated >1MiB responses -> opaque JSON error; now
  a clear "response exceeded N bytes" via readCapped.
- F12 step observer captured the caller ctx; now the merged runCtx.
- F13 compaction onFire was nil (doc claimed it logged); now wired to
  audit LogEvent("compaction_fired").
- F11 (no pre-dispatch hook in majordomo) documented honestly as a known
  limitation; F18 UsageSink doc clarified cache tokens are subsets of input.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 02:02:21 +00:00
steve dfbc5a42b9 P2: run.Executor — executus is runnable
The capstone of the run kernel: run.Executor.Run(ctx, RunnableAgent, inv)
ties model resolution + the tool registry + majordomo's agent loop +
context compaction + run-bounding + step/audit instrumentation into one
path, with every host concern behind the nil-safe run.Ports.

- run/executor.go: New(Config{Registry, Models, Defaults, Ports, Compactor,
  ContextTokens, SystemHeader}) + Run -> Result{RunID, Output, Steps, Usage,
  Err}. Budget gate (pre-run), model resolve, Audit StartRun/recorder
  (satisfies RunTally, stamped on inv.RunState), toolbox build, step observer
  (zips tool calls/results -> emitter + recorder.OnStep/OnTool), V10
  detached-MaxRuntime context with caller-cancel merged back, compaction wired
  from ContextTokens×ratio, audit Close + Budget Commit on a detached cleanup
  ctx. Zero Ports = a bounded in-memory run (gadfly's case).
- run/executor_test.go: hermetic end-to-end run against majordomo's fake
  provider (hello-world), Budget-rejection (no model call), Audit-port wiring
  (StartRun + Close with terminal status/output). All green under -race.
- examples/minimal upgraded to the real "hello, agentic world" (~15 lines:
  Configure tiers -> run.New -> Run -> print). README/CLAUDE.md updated.

Remaining P2 follow-ups (incremental): wire Critic/Checkpointer/PaletteSource/
Delivery into the loop, multi-phase Pipelines, and the no-tools direct path.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 02:02:21 +00:00
steve 130c2bdfab P2: move compactor -> compact/ + step instrumentation -> run/steps.go
- compact/compactor.go: the per-run stateful context compactor (token-threshold
  gate, fast-tier middle summarisation, fold memory) lifted from mort's
  skillexec/compactor.go. Self-contained; its only dependency is a ModelResolver
  func (model.ParseModelForContext satisfies it) + a token threshold.
- run/steps.go: the step-emission/instrumentation (stepEmitter, tool->kind/
  summary mapping with redaction, Result.Steps accumulation) from agentexec,
  repointed onto executus/tool.

Both build green. With the run-loop mechanics, RunnableAgent DTO, run.Ports,
compactor, and step instrumentation now all in place, the remaining P2 work is
the run.Executor itself (wiring these + majordomo's agent loop), which makes
executus runnable.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 02:02:21 +00:00
steve d9b44387f5 fix: address gadfly P1 review (3 low-risk findings)
Triaged gadfly's P1 review (advisory). Fixed the three clearly-correct,
low-risk items; the rest were pre-existing mort behavior or theoretical:

- model/call.go: recordUsage dropped fully-cached responses (input==0 &&
  output==0 early-return missed CacheRead/CacheWrite-only usage, which
  Anthropic/OpenAI prompt-caching bills). Guard now also checks cache tokens.
- llmmeta/helper.go: recordLedger swallowed Storage.RecordMetaCall errors;
  now logs them (slog.Warn) so a non-logging Storage impl can't silently drop
  audit rows.
- model/cloud_sync.go: the ollama.com limit-cache used unbounded io.ReadAll;
  wrapped both reads in io.LimitReader(1 MiB) so a misbehaving endpoint can't
  exhaust memory before the 15s timeout.

Noted-not-fixed (follow-ups / pre-existing mort semantics): tier_not_allowed
ledger label on resolution failure, unknown-model usage attribution, the
cloud_sync https scheme allowlist, and several theoretical/cosmetic items.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 02:02:21 +00:00
steve e0e2c0451a ci: sync gadfly review config to mort's foreman-provider setup
Mirror mort's updated adversarial-review.yml: m1/m5 pulled in via the
GADFLY_ENDPOINT_M1/_M5 secrets using gadfly's "foreman" provider type
(providers m1/m5; models m1/qwen3:14b, m5/qwen3.6:35b-mlx), 2 cloud models,
3-lens suite, pinned to the gadfly :sha-6e3a83c image. Header adjusted for
executus; functional config identical to mort's tested version.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 02:02:21 +00:00
steve c732a677df P2: define nil-safe run.Ports (the inversion spine)
Add run/ports.go: the host seams the executor will consume, every one
nil-safe so a light host runs with the zero Ports (no persistence/audit/
budget/critic/delegation/delivery) and a heavy host wires each to a battery.

Ports mirror mort's existing interfaces so the batteries implement them
directly:
- Audit + RunRecorder (mort skillaudit.Storage/Writer): StartRun -> per-run
  recorder (OnStep/OnTool/LogEvent/Close), recorder satisfies RunTally.
- Budget (mort skillexec.BudgetTracker): Check / Commit.
- Critic + CriticHandle (mort agentcritic): Monitor -> handle with
  RecordStep/RecordToolStart/Steer/Deadline/Stop (the loop wiring finalizes
  with the executor merge).
- Checkpointer (mort agentexec.RunCheckpointer): Save/Complete/Fail.
- PaletteSource (mort SkillInvokerForPalette + AgentInvokerForPalette):
  Resolve/Invoke skill + agent delegation.
Plus host-neutral RunInfo / RunStats.

This completes the P2 inversion DESIGN; the agentexec+skillexec ->
run.Executor merge that consumes these Ports is the remaining P2 work.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 02:02:21 +00:00
steve fa644f1826 P2 (foundation): run-loop mechanics + RunnableAgent DTO
Stand up the executus/run kernel foundation, decoupled from mort:

- runengine.go: the shared run-loop scaffolding (MergeCancellation,
  CleanupContextTimeout, RunFinalizer/FireFinalizers, RunStateAccessor) moved
  from mort. The accessor's *skillaudit.Writer dependency is inverted to a
  narrow run.RunTally interface (TokenStats + ToolCallsCount) — the kernel
  reads live tallies without importing the audit battery.
- submit.go: the legacy submit-capture compat tool (stdlib + majordomo/llm).
- agent.go: RunnableAgent DTO — the kernel's view of "a thing to run" (tier,
  prompt, caps, palette, phases, critic config). The persona Agent and saved
  Skill will LOWER into this DTO so the kernel never imports a noun battery.
  This is the spine of the agentexec.Run(*agents.Agent) inversion.

run/ builds with only majordomo + executus/tool. The executor merge
(agentexec+skillexec -> run.Executor) and the nil-safe run.Ports
(Audit/Critic/Budget/Checkpointer/PaletteSource) are the next P2 block.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 02:02:21 +00:00
28 changed files with 3828 additions and 43 deletions
+15 -17
View File
@@ -1,10 +1,11 @@
# 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
# merge. This reviews executus PRs (same setup as mort: m1/m5 locals + 2 cloud,
# 3-lens suite). Gadfly is a simple system — treat its findings as advisory and
# double-check before acting.
# merge. This reviews executus PRs with 3 ollama-cloud models (3-lens suite). Gadfly
# is a simple system — findings are advisory; always double-check before acting.
name: Adversarial Review (Gadfly)
@@ -40,24 +41,21 @@ jobs:
|| github.actor == 'fizi'
|| github.actor == 'dazed'))
runs-on: ubuntu-latest
# Full fleet (2 cloud + 2 local Macs, all running concurrently) reviewing
# every PR with the 3-lens suite — the slow local lanes dominate wall time.
timeout-minutes: 90
# 3 cloud models, all concurrent, 3-lens suite. ~12 min typical.
timeout-minutes: 30
steps:
- uses: docker://gitea.stevedudenhoeffer.com/steve/gadfly:v1
- uses: docker://gitea.stevedudenhoeffer.com/steve/gadfly:sha-6e3a83c
env:
GITEA_API: ${{ github.server_url }}/api/v1/repos/${{ github.repository }}
GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
OLLAMA_CLOUD_API_KEY: ${{ secrets.OLLAMA_CLOUD_API_KEY }}
# Local Ollama boxes (each its own lane, cap 1). NOTE: both Macs must be
# awake/reachable for their reviews to run; if a box is offline, that
# model's comment shows an error and the others still post.
GADFLY_ENDPOINT_M1PRO: "ollama|http://192.168.0.175:11434"
GADFLY_ENDPOINT_M5MAX: "ollama|http://192.168.0.173:11434"
# 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"
# 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"
# executus uses CLOUD MODELS ONLY. The local Macs (m1/m5) were dropped:
# on a P2-review measurement they took 2629 min (with lens timeouts)
# and contributed ZERO real findings — the two cloud models found every
# genuine bug in 612 min. Cloud-only is faster AND higher-signal.
# 3 cloud models, one consolidated comment each, all run in parallel.
GADFLY_MODELS: "minimax-m3:cloud,deepseek-v4-flash:cloud,glm-5.2:cloud"
GADFLY_PROVIDER_CONCURRENCY: "ollama-cloud=3"
# Default => the 3-lens suite (security, correctness, error-handling).
# Set the repo var GADFLY_SPECIALISTS to override (csv / "all" / "auto").
GADFLY_SPECIALISTS: ${{ vars.GADFLY_SPECIALISTS || 'security,correctness,error-handling' }}
+15 -4
View File
@@ -43,8 +43,13 @@ CORE (majordomo + stdlib):
fanout/ programmatic N×M swarm [P0 ✓]
deliver/ output egress seam (+ Discard/Stdout) [P0 ✓]
identity/ caller identity seams [P0 ✓]
run/ progress bridge now; the executor kernel + [P0 partial]
nil-safe Ports + RunnableAgent later [P2]
run/ run.Executor is RUNNABLE: model-resolve + [P2 core ✓]
toolbox + majordomo loop + compaction +
run-bounding (V10 detached timeout) + step/
audit observers + Budget gate; RunnableAgent
DTO + nil-safe run.Ports. Follow-ups: wire
Critic/Checkpointer/PaletteSource/Delivery,
Phases, and the no-tools direct path [P2]
dispatchguard/ loop/depth/fan-out caps [P0 ✓]
pendingattach/ attachment dedupe [P0 ✓]
tool/ registry + 3-stage permissions + ssrf [P1 ✓]
@@ -52,8 +57,14 @@ CORE (majordomo + stdlib):
(convar->config.Source; UsageSink/TraceSink seams; GenerateWith[T]
structured output — no separate structured/ pkg)
llmmeta/ shared meta-LLM helper over model/ [P1 ✓]
compact/ context compactor (WithCompactor hook) [P2]
tools/{web,net,store,compose,meta,comms} generic tools [P3]
compact/ context compactor (WithCompactor hook) [P2]
tools/ generic tool library: Register (think/now/ [P3 wip]
cite, zero-config) + RegisterMeta (classify/
extract_entities/summarize); seams in
research_providers.go/file_storage.go;
in-memory budget default. End-to-end "agent
calls a tool" test green. Remaining: web/net/
store/compose groups + their backends [P3]
BATTERIES (opt-in siblings, each nil-safe + a default):
persona/ Agent noun + AgentStore seam + yml loader [P4]
+13 -5
View File
@@ -31,15 +31,23 @@ bot) — mort and gadfly are the first two consumers (heavy and light). See
[mort]: https://gitea.stevedudenhoeffer.com/steve/mort
**Available today (P0):**
**Available today:**
- `run/`**executus is runnable.** `run.Executor` ties model resolution, the
tool registry, majordomo's agent loop, context compaction, run-bounding, and
step/audit instrumentation into one `Run(ctx, RunnableAgent, inv) Result`, with
every host concern behind a nil-safe `run.Ports` (Audit/Budget/Critic/
Checkpointer/PaletteSource/Delivery). See `examples/minimal`.
- `model/` — config-driven tier resolution + failover over majordomo, with
pluggable `UsageSink`/`TraceSink` and `GenerateWith[T]` structured output.
- `tool/` — the tool registry + 3-stage permission model + SSRF guard.
- `compact/` — the per-run context compactor.
- `lane/` — bounded worker pool with fair-share queueing (run- and
provider-concurrency).
- `fanout/` — programmatic N×M swarm with bounded global + per-key concurrency.
- `config/` — the host config seam (`Source`) with an env-var default.
- `deliver/` — the output-egress seam with `Discard`/`Stdout` defaults.
- `identity/` — caller-identity seams (`AdminPolicy`, `MemberResolver`).
- `dispatchguard/`, `pendingattach/`, `run/progress.go` — run-safety primitives.
- `config/`, `deliver/`, `identity/` — host seams (config / output / identity),
each with a shipped default.
- `dispatchguard/`, `pendingattach/` — run-safety primitives.
## Design
+369
View File
@@ -0,0 +1,369 @@
// V15.2 context compactor (re-based on majordomo).
//
// Why: the agent loop accumulates tool results indefinitely. A
// research-heavy run with many web_search / read_page / http_get
// results easily crosses 200K tokens and trips the model's HTTP-400
// "prompt too long" rejection mid-run (observed at 410K tokens on
// qwen3-coder:480b which has a 262K cap). majordomo's agent loop calls
// the compactor with the full message slice before every model call;
// the compactor returns a shorter slice that preserves the system
// prompt + recent messages, with the middle range replaced by a
// synthetic summary.
//
// Strategy (unchanged from the agentkit era):
// - Keep any leading system message verbatim. (Under majordomo the
// system prompt normally travels in Request.System, not in the
// message slice, so this is defensive.)
// - Keep the last KeepRecent messages verbatim. This ensures the
// agent has fresh tool state to act on; compacting too aggressively
// would strip the in-flight context it needs to make the next
// decision.
// - Compress the middle range via a single fast-tier LLM call that
// receives the middle messages as raw text and produces a paragraph
// summary (URLs visited, key findings, file_ids created, what the
// agent is trying to accomplish).
// - Replace the middle range with one synthetic user-role message
// containing the summary. (user-role chosen because tool-result-role
// would be ambiguous without a matching tool_call_id.)
//
// What moved in the majordomo conversion:
// - The token-threshold gate lives HERE now. agentkit estimated
// tokens and only invoked the compactor past a configured
// threshold; majordomo's hook fires before every model call, so
// the threshold check (estimateTokens vs the per-run threshold the
// executor computes from the model's context limit) is the
// compactor's first step.
// - The compactor is per-run STATEFUL: majordomo does not replace
// the loop's internal transcript with the compacted slice (the
// hook shapes only what is SENT), so without memory the middle
// would be re-summarised from scratch on every step past the
// threshold. The state remembers how far the transcript has been
// folded into the running summary and folds the previous summary
// into the next one instead of re-paying for it.
//
// Failure path: any error (LLM unavailable, malformed response, etc.)
// is returned to the agent loop, which treats compactor errors as
// non-fatal and sends the original slice — if the next model call hits
// the provider's limit, the existing HTTP-400 error path takes over.
package compact
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
)
// ModelResolver resolves a tier/spec to a usable llm.Model (and an enriched
// context for usage attribution). model.ParseModelForContext satisfies it.
type ModelResolver func(ctx context.Context, tier string) (context.Context, llm.Model, error)
// Compactor is the per-run compaction hook handed to the agent loop
// (matches the signature agent.WithCompactor expects).
type Compactor = func(ctx context.Context, msgs []llm.Message) ([]llm.Message, error)
// CompactionEvent describes one fired compaction so the executor can
// log it to skill_run_logs ("compaction_fired").
type CompactionEvent struct {
// MessagesBefore/After count the messages that would have been
// sent without/with this compaction.
MessagesBefore int
MessagesAfter int
// TokensBefore/After are the estimateTokens values for the same
// two slices.
TokensBefore int
TokensAfter int
}
// CompactorFactory mints a fresh per-run Compactor bound to a token
// threshold. onFire (nil-safe) observes every compaction that actually
// fires. A non-positive threshold yields a pass-through compactor.
type CompactorFactory func(thresholdTokens int, onFire func(CompactionEvent)) Compactor
// CompactorConfig controls the v15.2 compactor's behaviour. Construct
// in mort.go from convars + the executor's ModelResolver.
type CompactorConfig struct {
// Models resolves a model spec (typically "fast" or a specific tier)
// to an llm.Model. Required; nil disables compaction.
Models ModelResolver
// SummarizerTier is the model tier used for the compression LLM call.
// Production default "fast"; an admin may set "haiku" or a specific
// model spec. Empty falls back to "fast".
SummarizerTier string
// KeepRecent is the number of trailing messages preserved verbatim.
// Default 8. Lower values compact more aggressively (the next loop
// iteration sees less recent context); higher values keep more.
KeepRecent int
// SummaryWordCap bounds the LLM-generated summary length. Default
// 200 words ≈ 800 chars ≈ 200 tokens — small enough that the
// compaction always shrinks the slice meaningfully.
SummaryWordCap int
}
// NewCompactor returns a CompactorFactory implementing the middle-range
// summarisation strategy. nil cfg.Models returns a factory of no-op
// compactors that always return the input unchanged (degrades to
// v15.1 behaviour).
func NewCompactor(cfg CompactorConfig) CompactorFactory {
if cfg.Models == nil {
return func(int, func(CompactionEvent)) Compactor {
return func(_ context.Context, msgs []llm.Message) ([]llm.Message, error) {
return msgs, nil
}
}
}
if cfg.SummarizerTier == "" {
cfg.SummarizerTier = "fast"
}
if cfg.KeepRecent <= 0 {
cfg.KeepRecent = 8
}
if cfg.SummaryWordCap <= 0 {
cfg.SummaryWordCap = 200
}
return func(threshold int, onFire func(CompactionEvent)) Compactor {
st := &compactionState{}
return func(ctx context.Context, msgs []llm.Message) ([]llm.Message, error) {
return compactIfNeeded(ctx, cfg, st, threshold, onFire, msgs)
}
}
}
// compactionState is the per-run fold memory: msgs[:prefixEnd]
// (excluding a leading system message) are represented by summaryText
// in the rendered slice. Guarded by mu for safety although the agent
// loop invokes the hook from a single goroutine.
type compactionState struct {
mu sync.Mutex
prefixEnd int
summaryText string
}
// compactIfNeeded is the workhorse: render the transcript with any
// existing fold applied, check the threshold, and fold more of the
// middle into the summary when the rendered size still exceeds it.
func compactIfNeeded(ctx context.Context, cfg CompactorConfig, st *compactionState,
threshold int, onFire func(CompactionEvent), msgs []llm.Message) ([]llm.Message, error) {
st.mu.Lock()
defer st.mu.Unlock()
rendered := renderCompacted(st, msgs)
if threshold <= 0 {
return rendered, nil
}
tokensBefore := estimateTokens(rendered)
if tokensBefore < threshold {
return rendered, nil
}
// Determine the new middle range to fold: everything between what
// is already summarised (or the optional leading system message)
// and the KeepRecent tail.
startMiddle := st.prefixEnd
if startMiddle == 0 && len(msgs) > 0 && msgs[0].Role == llm.RoleSystem {
startMiddle = 1
}
endMiddle := len(msgs) - cfg.KeepRecent
if endMiddle <= startMiddle {
// Nothing new to fold (the tail alone exceeds the threshold).
// Return the rendered slice; the model call may still succeed.
return rendered, nil
}
middle := msgs[startMiddle:endMiddle]
summary, err := summariseMiddle(ctx, cfg, st.summaryText, middle)
if err != nil {
// Non-fatal upstream: the agent loop sends the ORIGINAL slice. Return
// msgs, not `rendered` — on a second+ compaction `rendered` already
// carries a prior synthetic summary, which is not the documented
// "original slice" the loop expects on a compactor error.
return msgs, fmt.Errorf("compactor: summarise middle: %w", err)
}
st.summaryText = summary
st.prefixEnd = endMiddle
out := renderCompacted(st, msgs)
if onFire != nil {
onFire(CompactionEvent{
MessagesBefore: len(rendered),
MessagesAfter: len(out),
TokensBefore: tokensBefore,
TokensAfter: estimateTokens(out),
})
}
return out, nil
}
// renderCompacted applies the fold state to msgs: [optional system] +
// [synthetic summary] + msgs[prefixEnd:]. With no fold yet, msgs is
// returned unchanged.
func renderCompacted(st *compactionState, msgs []llm.Message) []llm.Message {
if st.prefixEnd <= 0 || st.prefixEnd > len(msgs) {
return msgs
}
tail := msgs[st.prefixEnd:]
out := make([]llm.Message, 0, len(tail)+2)
if msgs[0].Role == llm.RoleSystem {
out = append(out, msgs[0])
}
out = append(out, llm.UserText(
"[CONTEXT COMPACTED] The earlier portion of this conversation was summarised "+
"to fit the model's context window. Summary:\n\n"+st.summaryText+
"\n\nResume from the recent messages below."))
out = append(out, tail...)
return out
}
// summariseMiddle composes a "compress this transcript" prompt and
// fires one fast-tier LLM call. prevSummary (may be empty) is the
// running summary from earlier compactions; it is folded into the new
// summary so prior context is not lost.
func summariseMiddle(ctx context.Context, cfg CompactorConfig, prevSummary string, middle []llm.Message) (string, error) {
if len(middle) == 0 {
return "", errors.New("compactor: empty middle range")
}
modelCtx, model, err := cfg.Models(ctx, cfg.SummarizerTier)
if err != nil {
return "", fmt.Errorf("compactor: resolve summarizer model %q: %w", cfg.SummarizerTier, err)
}
if model == nil {
return "", errors.New("compactor: summarizer model resolved to nil")
}
if modelCtx != nil {
ctx = modelCtx
}
var prior string
if strings.TrimSpace(prevSummary) != "" {
prior = "AN EARLIER PORTION WAS ALREADY SUMMARISED AS:\n" + prevSummary +
"\n\nFold that summary into your new one — its facts must survive.\n\n"
}
transcript := renderTranscript(middle)
prompt := fmt.Sprintf(
"You are compressing an in-flight agent's conversation transcript so the agent "+
"can continue working without blowing its model context. The transcript below is "+
"a sequence of tool calls and their results. Produce a single paragraph (under %d words) "+
"that captures:\n"+
" - WHAT the agent has been trying to accomplish.\n"+
" - WHICH URLs were visited / fetched (list inline, comma-separated).\n"+
" - KEY findings or decisions (factual results the agent will need later).\n"+
" - ANY file_ids or KV keys the agent created — these are persistent state references the agent must keep.\n"+
" - ANY errors or dead-ends that the agent should not re-try.\n"+
"DO NOT include verbose HTTP headers, tool-call metadata, error stack traces, or repetitive content. "+
"DO NOT add commentary or markdown headers. Output prose only.\n\n"+
"%sTRANSCRIPT TO COMPRESS:\n%s",
cfg.SummaryWordCap,
prior,
transcript,
)
resp, err := model.Generate(ctx, llm.Request{Messages: []llm.Message{llm.UserText(prompt)}})
if err != nil {
return "", fmt.Errorf("compactor: summarise LLM call: %w", err)
}
text := strings.TrimSpace(resp.Text())
if text == "" {
return "", errors.New("compactor: summarizer returned empty text")
}
return text, nil
}
// estimateTokens is the chars/4 heuristic over a message slice's text,
// tool calls, and tool results. Images count a flat ~1K tokens each.
// It intentionally matches the coarse estimator the old agentkit loop
// used — the 0.7 threshold ratio provides the safety margin.
func estimateTokens(msgs []llm.Message) int {
chars := 0
for _, m := range msgs {
for _, p := range m.Parts {
switch v := p.(type) {
case llm.TextPart:
chars += len(v.Text)
case llm.ImagePart:
chars += 4096
default:
// llm.Part is a sealed-but-extensible interface (future media
// kinds). Count an unknown part conservatively (like an image)
// rather than 0, so a transcript of unrecognised content can't
// silently slip under the compaction threshold and 400 the
// model. Bump this if a large new part kind lands.
_ = v
chars += 4096
}
}
for _, tc := range m.ToolCalls {
chars += len(tc.Name) + len(tc.Arguments)
}
for _, tr := range m.ToolResults {
chars += len(tr.Content)
}
}
return chars / 4
}
// transcriptMessageCap bounds individual message bodies at ~2KB so a
// single ultra-long tool result can't dominate the prompt sent to the
// summarizer.
const transcriptMessageCap = 2048
// secretBearingTools name tools whose ARGUMENTS routinely carry credentials or
// message bodies (bearer tokens, API keys, recipients, request bodies). Their
// args are dropped before the transcript reaches the summarizer model — which
// may be a different provider/tier than the run model — mirroring the redaction
// run/steps.go applies to user-facing step summaries. http_* and webhook_* are
// matched by prefix below.
var secretBearingTools = map[string]bool{
"mcp_call": true,
"email_send": true,
}
// redactToolArgs returns a summariser-safe rendering of a tool call's args:
// "[redacted]" for known secret-bearing tools, the args verbatim otherwise.
func redactToolArgs(name, args string) string {
if secretBearingTools[name] ||
strings.HasPrefix(name, "http_") ||
strings.HasPrefix(name, "webhook_") {
return "[redacted]"
}
return args
}
// renderTranscript flattens a message slice to a plain-text transcript
// suitable for the summarisation prompt. Tool calls show name + (redacted) args,
// tool results show name + body. Empty fields are skipped.
//
// NOTE: tool-RESULT bodies are forwarded to the summarizer (the summary needs
// the findings). A host whose tool results may contain secrets and whose
// summarizer tier resolves to an untrusted provider should ensure that tier is
// trusted, or pre-sanitise results before they reach the agent loop.
func renderTranscript(msgs []llm.Message) string {
var sb strings.Builder
for i, m := range msgs {
fmt.Fprintf(&sb, "---\n[%d] role=%s\n", i+1, m.Role)
if text := m.Text(); text != "" {
sb.WriteString(truncate(text, transcriptMessageCap))
sb.WriteString("\n")
}
for _, tc := range m.ToolCalls {
fmt.Fprintf(&sb, "tool_call name=%s args=%s\n", tc.Name, truncate(redactToolArgs(tc.Name, string(tc.Arguments)), transcriptMessageCap))
}
for _, tr := range m.ToolResults {
fmt.Fprintf(&sb, "tool_result name=%s body=%s\n", tr.Name, truncate(tr.Content, transcriptMessageCap))
}
}
return sb.String()
}
func truncate(s string, n int) string {
if len(s) <= n {
return s
}
return s[:n] + "...(truncated)"
}
+35 -13
View File
@@ -1,27 +1,49 @@
// Command minimal demonstrates executus's standalone core primitives available
// today (P0): the config seam + bounded fan-out. The full zero-config "agentic
// in ~12 lines" example arrives once the model, tool, and run packages land
// (P1P3).
// Command minimal is executus's "hello, agentic world": wire a model resolver,
// a tool registry, and the run executor, then run an agent. With no batteries
// (Audit/Budget/Critic/Checkpointer/Palette/Delivery all nil) this is a
// bounded, in-memory run — the light-host shape (gadfly's case).
//
// Run it with a provider key for the configured tier, e.g.
//
// ANTHROPIC_API_KEY=sk-... go run ./examples/minimal
//
// Override a tier from the environment without touching code, e.g.
//
// EXECUTUS_MODEL_TIER_FAST=openai/gpt-4o-mini ANTHROPIC_API_KEY= OPENAI_KEY=sk-... go run ./examples/minimal
package main
import (
"context"
"fmt"
"log"
"gitea.stevedudenhoeffer.com/steve/executus/config"
"gitea.stevedudenhoeffer.com/steve/executus/fanout"
"gitea.stevedudenhoeffer.com/steve/executus/model"
"gitea.stevedudenhoeffer.com/steve/executus/run"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
func main() {
cfg := config.Env("EXECUTUS_") // e.g. EXECUTUS_FANOUT_MAX_CONCURRENT=8
max := cfg.Int("fanout.max_concurrent", 4)
// 1. Configure model tiers: live values come from the environment
// (EXECUTUS_MODEL_TIER_<NAME>), falling back to these defaults.
model.Configure(config.Env("EXECUTUS_"), map[string]string{
"fast": "anthropic/claude-haiku-4-5",
"thinking": "anthropic/claude-opus-4-8",
}, 0)
items := []string{"alpha", "beta", "gamma", "delta"}
results := fanout.Run(context.Background(), items,
fanout.Options[string]{MaxConcurrent: max},
func(_ context.Context, s string) (int, error) { return len(s), nil })
// 2. Build the executor: a tool registry + the model resolver. No batteries.
ex := run.New(run.Config{
Registry: tool.NewRegistry(),
Models: model.ParseModelForContext,
})
for _, r := range results {
fmt.Printf("%-6s -> %d (err=%v)\n", items[r.Index], r.Value, r.Err)
// 3. Run an agent and print its answer.
res := ex.Run(context.Background(),
run.RunnableAgent{Name: "assistant", SystemPrompt: "You are concise.", ModelTier: "fast"},
tool.Invocation{RunID: "demo-1", CallerID: "local"},
"In one sentence, what is an agent harness?")
if res.Err != nil {
log.Fatalf("run failed: %v", res.Err)
}
fmt.Println(res.Output)
}
+4 -1
View File
@@ -34,6 +34,7 @@ import (
"context"
"encoding/json"
"fmt"
"log/slog"
"strings"
"time"
@@ -574,7 +575,9 @@ func (h *Helper) recordLedger(ctx context.Context, call MetaCall) {
if h.storage == nil {
return
}
_ = h.storage.RecordMetaCall(ctx, call)
if err := h.storage.RecordMetaCall(ctx, call); err != nil {
slog.Warn("llmmeta: failed to record ledger row", "err", err)
}
}
// tryParseJSON attempts to decode text as JSON. Returns the parsed
+1 -1
View File
@@ -237,7 +237,7 @@ func recordUsage(ctx context.Context, resp *llm.Response) {
return
}
u := resp.Usage
if u.InputTokens == 0 && u.OutputTokens == 0 {
if u.InputTokens == 0 && u.OutputTokens == 0 && u.CacheReadTokens == 0 && u.CacheWriteTokens == 0 {
return
}
model := resolvedModelName(ctx, resp)
+23 -2
View File
@@ -314,7 +314,7 @@ func (c *CloudOllamaLimitCache) fetchTags(ctx context.Context) ([]string, error)
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
body, err := readCapped(resp.Body)
if err != nil {
return nil, err
}
@@ -367,7 +367,7 @@ func (c *CloudOllamaLimitCache) fetchContextLength(ctx context.Context, modelNam
return 0, err
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
respBody, err := readCapped(resp.Body)
if err != nil {
return 0, err
}
@@ -451,3 +451,24 @@ func truncate(b []byte, n int) string {
}
return string(b[:n]) + "...(truncated)"
}
// maxLimitCacheResponseBytes bounds the ollama.com limit-cache HTTP responses
// (/api/tags, /api/show) so a misbehaving endpoint can't stream an unbounded
// body before the 15s timeout fires. 1 MiB is far above any real response.
const maxLimitCacheResponseBytes = 1 << 20
// readCapped reads up to maxLimitCacheResponseBytes from r and returns a clear
// error if the response EXCEEDS the cap — rather than silently truncating (as a
// bare io.LimitReader does) and letting downstream json.Unmarshal fail with an
// opaque "unexpected end of JSON input". It reads one extra byte to detect the
// overflow.
func readCapped(r io.Reader) ([]byte, error) {
body, err := io.ReadAll(io.LimitReader(r, maxLimitCacheResponseBytes+1))
if err != nil {
return nil, err
}
if len(body) > maxLimitCacheResponseBytes {
return nil, fmt.Errorf("cloud_sync: response exceeded %d bytes", maxLimitCacheResponseBytes)
}
return body, nil
}
+5
View File
@@ -16,6 +16,11 @@ import (
// UsageSink receives one record per successful Generate through a model parsed
// by this package (ParseModelRequest / ParseModelForContext). Implement it to
// meter or bill; the token detail mirrors majordomo's Response.Usage.
//
// IMPORTANT: cacheReadTokens and cacheWriteTokens are PORTIONS of inputTokens,
// not independent additive values (they let a sink price cached vs fresh input
// differently). A sink must NOT compute total = input+output+cacheRead+
// cacheWrite — that double-counts the cached input.
type UsageSink interface {
Record(ctx context.Context, model string, inputTokens, outputTokens, cacheReadTokens, cacheWriteTokens int)
}
+76
View File
@@ -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
}
+318
View File
@@ -0,0 +1,318 @@
package run
import (
"context"
"errors"
"fmt"
"time"
"gitea.stevedudenhoeffer.com/steve/majordomo/agent"
"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
"gitea.stevedudenhoeffer.com/steve/executus/compact"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// ModelResolver resolves a tier alias or concrete spec to a usable llm.Model
// and an enriched context (for usage attribution). model.ParseModelForContext
// satisfies it.
type ModelResolver func(ctx context.Context, tier string) (context.Context, llm.Model, error)
// Defaults are the executor's fallback caps and loop guards, applied per run
// when the RunnableAgent leaves a field zero.
type Defaults struct {
MaxIterations int // tool-dispatch steps; default 12
MaxRuntime time.Duration // wall-clock per run; default 60s
FallbackTier string // tier when the agent's is empty; default "fast"
MaxConsecutiveToolErrors int // loop guard; default 3
MaxSameToolCallRepeats int // retry-storm guard; default 3
CompactionThresholdRatio float64 // fraction of model context to compact at; default 0.7
}
func (d Defaults) withFallbacks() Defaults {
if d.MaxIterations <= 0 {
d.MaxIterations = 12
}
if d.MaxRuntime <= 0 {
d.MaxRuntime = 60 * time.Second
}
if d.FallbackTier == "" {
d.FallbackTier = "fast"
}
if d.MaxConsecutiveToolErrors <= 0 {
d.MaxConsecutiveToolErrors = 3
}
if d.MaxSameToolCallRepeats <= 0 {
d.MaxSameToolCallRepeats = 3
}
if d.CompactionThresholdRatio <= 0 {
d.CompactionThresholdRatio = 0.7
}
return d
}
// Config wires an Executor. Registry + Models are required; everything else is
// optional and nil-safe — the zero Config beyond those yields a bounded,
// in-memory run with no persistence/audit/budget/critic/delegation/compaction
// (gadfly's case).
type Config struct {
Registry tool.Registry
Models ModelResolver
Defaults Defaults
Ports Ports
// Compactor mints the per-run context-compaction hook. nil disables
// compaction. ContextTokens resolves a tier's model context-window (for
// the compaction threshold); nil — or a zero return — also disables it.
Compactor compact.CompactorFactory
ContextTokens func(tier string) int
// SystemHeader is an optional platform header prepended to every agent's
// system prompt.
SystemHeader string
}
// Executor runs a RunnableAgent through majordomo's agent loop with the wired
// Ports. Construct with New; safe for concurrent use across runs.
type Executor struct {
cfg Config
}
// New builds an Executor. It panics if Registry or Models is nil — those are
// structural, not runtime, errors.
func New(cfg Config) *Executor {
if cfg.Registry == nil || cfg.Models == nil {
panic("run.New: Registry and Models are required")
}
cfg.Defaults = cfg.Defaults.withFallbacks()
return &Executor{cfg: cfg}
}
// Result is one run's outcome. Err carries the run failure (if any); the other
// fields are populated best-effort even on error (partial output/steps/usage).
type Result struct {
RunID string
Output string
Steps []tool.Step
Usage llm.Usage
Err error
}
// Run executes ra with the given invocation + input and returns the Result. It
// never propagates a panic; failures surface in Result.Err.
func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocation, input string) Result {
started := time.Now()
res := Result{RunID: inv.RunID}
tier := ra.ModelTier
if tier == "" {
tier = e.cfg.Defaults.FallbackTier
}
maxIter := ra.MaxIterations
if maxIter <= 0 {
maxIter = e.cfg.Defaults.MaxIterations
}
maxRuntime := ra.MaxRuntime
if maxRuntime <= 0 {
maxRuntime = e.cfg.Defaults.MaxRuntime
}
// Budget gate (pre-run): a rejected run makes no model call.
if e.cfg.Ports.Budget != nil {
if err := e.cfg.Ports.Budget.Check(ctx, inv.CallerID); err != nil {
res.Err = err
return res
}
}
// Resolve the model (enriches ctx for usage attribution).
modelCtx, model, err := e.cfg.Models(ctx, tier)
if err != nil {
res.Err = fmt.Errorf("resolve model %q: %w", tier, err)
return res
}
if model == nil {
// A resolver returning (ctx, nil, nil) would otherwise nil-panic inside
// the agent loop; surface it as a clean error (Run never panics out).
res.Err = fmt.Errorf("resolve model %q: resolver returned a nil model", tier)
return res
}
ctx = modelCtx
// Audit start (optional). The recorder satisfies RunTally; stamp it on the
// invocation so a self-status tool can read live progress.
var rec RunRecorder
var stateAcc *RunStateAccessor
if e.cfg.Ports.Audit != nil {
rec = e.cfg.Ports.Audit.StartRun(ctx, RunInfo{
RunID: inv.RunID,
SubjectID: ra.ID,
Name: ra.Name,
CallerID: inv.CallerID,
ChannelID: inv.ChannelID,
ParentRunID: inv.ParentRunID,
Inputs: inv.SkillInputs,
StartedAt: started,
})
}
if rec != nil {
stateAcc = NewRunStateAccessor(rec, maxIter, 0, started)
inv.RunState = stateAcc
}
// Build the toolbox from the agent's low-level tools.
toolbox, err := e.cfg.Registry.Build(ra.LowLevelTools, inv, tool.Visibility("private"), nil)
if err != nil {
res.Err = fmt.Errorf("build toolbox: %w", err)
e.finishAudit(ctx, rec, "error", res, started, res.Err)
return res
}
// Run context: bound by MaxRuntime, detached from the caller's deadline so a
// lane/queue wait doesn't eat the run budget (mort's V10 lesson). Caller
// cancellation still propagates via MergeCancellation. Created BEFORE the
// step observer so the observer forwards the merged run context (not a
// possibly-cancelled caller ctx) to OnStep consumers.
runCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), maxRuntime)
defer cancel()
runCtx, mergeCancel := MergeCancellation(runCtx, ctx)
defer mergeCancel()
// Step instrumentation: accumulate Result.Steps + fire inv.OnStep, feed the
// audit recorder, and keep the live iteration counter fresh. majordomo's
// step observer hands us each completed iteration; we zip the model's tool
// calls with their executed results PAIRWISE — a result without a matching
// call (or a call without a result) is skipped rather than recorded as an
// empty-name "ghost" step.
emitter := newStepEmitter(inv.OnStep)
stepObserver := func(s agent.Step) {
if stateAcc != nil {
stateAcc.SetIteration(s.Index)
}
if rec != nil {
rec.OnStep(s.Index, s.Response)
}
var calls []llm.ToolCall
if s.Response != nil {
calls = s.Response.ToolCalls
}
n := len(s.Results)
if len(calls) < n {
n = len(calls)
}
for i := 0; i < n; i++ {
call, r := calls[i], s.Results[i]
emitter.toolStart(runCtx, call.Name, call.Arguments)
emitter.toolEnd(runCtx, call, r.Content, r.IsError)
if rec != nil {
rec.OnTool(call, r.Content)
}
}
}
opts := []agent.Option{
agent.WithToolbox(toolbox),
agent.WithMaxSteps(maxIter),
agent.WithToolErrorLimits(e.cfg.Defaults.MaxConsecutiveToolErrors, e.cfg.Defaults.MaxSameToolCallRepeats),
agent.WithStepObserver(stepObserver),
}
if e.cfg.Compactor != nil && e.cfg.ContextTokens != nil {
if threshold := e.compactionThreshold(tier); threshold > 0 {
// Forward compaction events to the audit log (makes the
// CompactionEvent doc's "logged to the run trace" promise true).
var onFire func(compact.CompactionEvent)
if rec != nil {
onFire = func(ev compact.CompactionEvent) {
rec.LogEvent("compaction_fired", map[string]any{
"messages_before": ev.MessagesBefore,
"messages_after": ev.MessagesAfter,
"tokens_before": ev.TokensBefore,
"tokens_after": ev.TokensAfter,
})
}
}
opts = append(opts, agent.WithCompactor(e.cfg.Compactor(threshold, onFire)))
}
}
ag := agent.New(model, e.systemPrompt(ra), opts...)
runRes, runErr := ag.Run(runCtx, input)
status := statusFor(runErr)
if runRes != nil {
res.Output = runRes.Output
res.Usage = runRes.Usage
}
res.Steps = emitter.snapshot()
res.Err = runErr
e.finishAudit(ctx, rec, status, res, started, runErr)
if e.cfg.Ports.Budget != nil {
e.cfg.Ports.Budget.Commit(detach(ctx), inv.CallerID, time.Since(started).Seconds())
}
return res
}
// statusFor maps a run error to a RunStats.Status, distinguishing a deadline
// (timeout) and a cancellation (cancelled — caller cancel or shutdown) from a
// generic error so audit consumers can tell them apart.
func statusFor(runErr error) string {
switch {
case runErr == nil:
return "ok"
case errors.Is(runErr, context.DeadlineExceeded):
return "timeout"
case errors.Is(runErr, context.Canceled):
return "cancelled"
default:
return "error"
}
}
// finishAudit writes the terminal roll-up on a detached context so a cancelled
// run still records (mort's CleanupContextTimeout lesson).
func (e *Executor) finishAudit(ctx context.Context, rec RunRecorder, status string, res Result, started time.Time, runErr error) {
if rec == nil {
return
}
stats := RunStats{
Status: status,
Output: res.Output,
ToolCalls: rec.ToolCallsCount(),
RuntimeSeconds: time.Since(started).Seconds(),
}
if runErr != nil {
stats.Error = runErr.Error()
}
stats.InputTokens, stats.OutputTokens, stats.ThinkingTokens = rec.TokenStats()
rec.Close(detach(ctx), stats)
}
func (e *Executor) systemPrompt(ra RunnableAgent) string {
if e.cfg.SystemHeader == "" {
return ra.SystemPrompt
}
if ra.SystemPrompt == "" {
return e.cfg.SystemHeader
}
return e.cfg.SystemHeader + "\n\n" + ra.SystemPrompt
}
// compactionThreshold returns the token threshold for the tier's model context
// window (ratio × limit), or 0 when the limit is unknown.
func (e *Executor) compactionThreshold(tier string) int {
max := e.cfg.ContextTokens(tier)
if max <= 0 {
return 0
}
return int(float64(max) * e.cfg.Defaults.CompactionThresholdRatio)
}
// detach derives a bounded cleanup context off ctx, detached from its
// cancellation, for post-run writes. The cancel is intentionally not returned;
// CleanupContextTimeout bounds the lifetime.
func detach(ctx context.Context) context.Context {
c, cancel := context.WithTimeout(context.WithoutCancel(ctx), CleanupContextTimeout)
_ = cancel // bounded by the timeout; nothing to cancel early
return c
}
+168
View File
@@ -0,0 +1,168 @@
package run
import (
"context"
"errors"
"fmt"
"testing"
"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
"gitea.stevedudenhoeffer.com/steve/majordomo/provider/fake"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// fakeModels returns a ModelResolver backed by a fake provider scripted to
// reply with the given text (no tool calls — the loop terminates immediately).
func fakeModels(t *testing.T, reply string) ModelResolver {
t.Helper()
fp := fake.New("fake")
fp.Enqueue("test-model", fake.Reply(reply))
m, err := fp.Model("test-model")
if err != nil {
t.Fatalf("fake model: %v", err)
}
return func(ctx context.Context, _ string) (context.Context, llm.Model, error) {
return ctx, m, nil
}
}
// TestExecutorRunHelloWorld is the milestone: executus runs an agent end-to-end
// against the fake provider and returns its output. Proves the kernel is
// runnable with the zero Ports (no persistence/audit/budget/critic).
func TestExecutorRunHelloWorld(t *testing.T) {
ex := New(Config{
Registry: tool.NewRegistry(),
Models: fakeModels(t, "hello from executus"),
})
res := ex.Run(context.Background(),
RunnableAgent{Name: "greeter", SystemPrompt: "be brief", ModelTier: "test-model"},
tool.Invocation{RunID: "run-1", CallerID: "caller-1"},
"say hi")
if res.Err != nil {
t.Fatalf("run error: %v", res.Err)
}
if res.Output != "hello from executus" {
t.Fatalf("output = %q, want %q", res.Output, "hello from executus")
}
if res.RunID != "run-1" {
t.Errorf("RunID = %q, want run-1", res.RunID)
}
}
// TestExecutorBudgetRejection: a Budget that denies makes no model call.
func TestExecutorBudgetRejection(t *testing.T) {
denied := errors.New("over budget")
var modelCalled bool
models := func(ctx context.Context, _ string) (context.Context, llm.Model, error) {
modelCalled = true
return ctx, nil, nil
}
ex := New(Config{
Registry: tool.NewRegistry(),
Models: models,
Ports: Ports{Budget: budgetFunc{check: func(string) error { return denied }}},
})
res := ex.Run(context.Background(),
RunnableAgent{ModelTier: "test-model"},
tool.Invocation{RunID: "r", CallerID: "broke"}, "hi")
if !errors.Is(res.Err, denied) {
t.Fatalf("err = %v, want budget denial", res.Err)
}
if modelCalled {
t.Error("model must not be resolved/called when budget denies")
}
}
// TestExecutorAuditWiring: the Audit port receives StartRun + Close with the
// terminal status/output.
func TestExecutorAuditWiring(t *testing.T) {
rec := &captureRecorder{}
ex := New(Config{
Registry: tool.NewRegistry(),
Models: fakeModels(t, "done"),
Ports: Ports{Audit: auditFunc{start: func(RunInfo) RunRecorder { return rec }}},
})
res := ex.Run(context.Background(),
RunnableAgent{ModelTier: "test-model"},
tool.Invocation{RunID: "r2", CallerID: "c"}, "go")
if res.Err != nil {
t.Fatalf("run error: %v", res.Err)
}
if !rec.closed {
t.Fatal("recorder.Close was not called")
}
if rec.stats.Status != "ok" {
t.Errorf("close status = %q, want ok", rec.stats.Status)
}
if rec.stats.Output != "done" {
t.Errorf("close output = %q, want done", rec.stats.Output)
}
}
// --- test doubles ---
type budgetFunc struct{ check func(callerID string) error }
func (b budgetFunc) Check(_ context.Context, callerID string) error { return b.check(callerID) }
func (b budgetFunc) Commit(context.Context, string, float64) {}
type auditFunc struct{ start func(RunInfo) RunRecorder }
func (a auditFunc) StartRun(_ context.Context, info RunInfo) RunRecorder { return a.start(info) }
type captureRecorder struct {
closed bool
stats RunStats
steps int
tools int
}
func (r *captureRecorder) TokenStats() (in, out, thinking int64) { return 0, 0, 0 }
func (r *captureRecorder) ToolCallsCount() int { return r.tools }
func (r *captureRecorder) OnStep(int, *llm.Response) { r.steps++ }
func (r *captureRecorder) OnTool(llm.ToolCall, string) { r.tools++ }
func (r *captureRecorder) LogEvent(string, map[string]any) {}
func (r *captureRecorder) LogError(string) {}
func (r *captureRecorder) Close(_ context.Context, s RunStats) { r.closed = true; r.stats = s }
// TestExecutorNilModelNoPanic: a resolver returning (ctx, nil, nil) yields a
// clean error, not a nil-pointer panic (gadfly F1, high severity).
func TestExecutorNilModelNoPanic(t *testing.T) {
ex := New(Config{
Registry: tool.NewRegistry(),
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) {
return ctx, nil, nil // nil model, nil error
},
})
res := ex.Run(context.Background(),
RunnableAgent{ModelTier: "x"}, tool.Invocation{RunID: "r"}, "hi")
if res.Err == nil {
t.Fatal("expected an error for a nil model, got nil (would have panicked in the loop)")
}
}
// TestStatusFor maps run errors to RunStats.Status (gadfly F3).
func TestStatusFor(t *testing.T) {
cases := []struct {
err error
want string
}{
{nil, "ok"},
{context.DeadlineExceeded, "timeout"},
{context.Canceled, "cancelled"},
{fmt.Errorf("wrapped: %w", context.DeadlineExceeded), "timeout"},
{errors.New("boom"), "error"},
}
for _, c := range cases {
if got := statusFor(c.err); got != c.want {
t.Errorf("statusFor(%v) = %q, want %q", c.err, got, c.want)
}
}
}
+168
View File
@@ -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)
}
+157
View File
@@ -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()),
}
}
+419
View File
@@ -0,0 +1,419 @@
package run
// steps.go — the per-run step emitter and the tool→step presentation
// mapping. This is the single place that turns the executor's post-step
// observer into ordered tool.Step records: one per tool call, each with a
// stable id, an open-vocabulary kind, and a human present-tense summary
// that flips running→complete/error.
//
// One source feeds two consumers (mirroring the OnEvent/OnToolEvent/
// PostRunResult pattern): the live tool.Invocation.OnStep callback
// (nil-safe) AND snapshot(), which the executor copies onto Result.Steps.
// Because the Result accumulation does not depend on OnStep being set,
// every surface — chat (JSON + SSE), Discord, cron, sub-agents — carries
// steps; OnStep is needed only for live streaming.
//
// LIMITATION (current): majordomo exposes only a POST-step observer, so
// the executor calls toolStart+toolEnd back-to-back after each tool has
// already run. Steps are therefore recorded faithfully, but step.StartedAt
// ≈ EndedAt and the intermediate "running" phase is never observable to a
// live OnStep consumer. A pre-dispatch hook (wrapping each tool's handler
// to emit toolStart before execution, like mort's state-react decorator)
// is a follow-up that would restore real start timing + the running phase.
// The emitter already supports that two-call shape — toolStart and toolEnd
// are separate methods — so wiring it later is additive.
import (
"context"
"encoding/json"
"fmt"
"net/url"
"strings"
"time"
"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// stepSummaryMaxLen caps the human summary length (section G size cap).
// Detail is unused in v1 (no live detail source while replies are
// generated blocking) so there is no Detail cap yet.
const stepSummaryMaxLen = 200
// stepEmitter accumulates ordered steps for one run and fires the live
// OnStep callback.
//
// Concurrency: touched ONLY from the agent-loop goroutine — the executor's
// stepObserver (and, once a pre-dispatch hook is wired, that hook) run
// there; majordomo executes a step's tool calls sequentially, and
// sub-agents build their own Invocation so they never reach this
// emitter. Same single-goroutine contract as the audit Writer and the
// critic ProgressRecorder — no internal lock.
type stepEmitter struct {
onStep func(ctx context.Context, ev tool.StepEvent)
now func() time.Time
seq int
steps []tool.Step // ordered; the snapshot for Result.Steps
byID map[string]int // step id -> index into steps
pending map[string][]string // correlation key -> queued running ids (FIFO)
}
// newStepEmitter returns an emitter that forwards to onStep (nil-safe).
func newStepEmitter(onStep func(ctx context.Context, ev tool.StepEvent)) *stepEmitter {
return &stepEmitter{
onStep: onStep,
now: time.Now,
byID: map[string]int{},
pending: map[string][]string{},
}
}
// corrKey correlates a "start" (name + raw args, no call id available at
// the pre-dispatch hook) with its later "end" (the stepObserver has the
// full call incl. id + the same raw args).
func corrKey(name string, args json.RawMessage) string {
return name + "\x00" + string(args)
}
// toolStart records + emits the "start" of a tool call. Called from the
// pre-dispatch hookToolbox closure, before the tool runs.
func (e *stepEmitter) toolStart(ctx context.Context, name string, args json.RawMessage) {
if e == nil {
return
}
step := e.newStep(name, args)
key := corrKey(name, args)
e.pending[key] = append(e.pending[key], step.ID)
e.fire(ctx, "start", step)
}
// toolEnd records + emits the terminal "end" of a tool call. Called from
// the stepObserver for each completed tool call. If no matching start was
// seen (e.g. a tool with a nil handler the pre-dispatch hook skipped), a
// start is synthesized so the step still appears.
func (e *stepEmitter) toolEnd(ctx context.Context, call llm.ToolCall, result string, isError bool) {
if e == nil {
return
}
id := e.matchPending(call.Name, call.Arguments)
if id == "" {
id = e.newStep(call.Name, call.Arguments).ID
}
idx, ok := e.byID[id]
if !ok {
return
}
step := &e.steps[idx]
end := e.now()
step.EndedAt = &end
if isError {
step.Status = "error"
} else {
step.Status = "complete"
}
if s := summaryForEnd(call.Name, call.Arguments, result, isError); s != "" {
step.Summary = s
}
e.fire(ctx, "end", *step)
}
// newStep mints + appends a running step and returns it (by value).
func (e *stepEmitter) newStep(name string, args json.RawMessage) tool.Step {
e.seq++
step := tool.Step{
ID: fmt.Sprintf("s%d", e.seq),
Kind: kindForTool(name),
Title: name,
Summary: summaryForStart(name, args),
Status: "running",
StartedAt: e.now(),
}
e.byID[step.ID] = len(e.steps)
e.steps = append(e.steps, step)
return step
}
// matchPending pops the oldest running step id for (name, args). Falls back to
// the OLDEST still-running step of the same tool name when the args don't
// byte-match between start and end (e.g. JSON key reordering). FIFO on the
// fallback too, consistent with the per-key queue pop above — closing the
// oldest avoids mis-correlating concurrent same-named calls. Returns "" on no
// match.
func (e *stepEmitter) matchPending(name string, args json.RawMessage) string {
key := corrKey(name, args)
if q := e.pending[key]; len(q) > 0 {
id := q[0]
if len(q) == 1 {
delete(e.pending, key)
} else {
e.pending[key] = q[1:]
}
return id
}
for i := 0; i < len(e.steps); i++ {
if e.steps[i].Title == name && e.steps[i].Status == "running" {
return e.steps[i].ID
}
}
return ""
}
func (e *stepEmitter) fire(ctx context.Context, phase string, step tool.Step) {
if e.onStep == nil {
return
}
e.onStep(ctx, tool.StepEvent{Phase: phase, Step: step})
}
// snapshot returns a copy of the ordered, deduplicated step set for the
// run Result. A step still "running" at run end (e.g. the run was
// cancelled mid-tool-call) is reported as-is.
func (e *stepEmitter) snapshot() []tool.Step {
if e == nil || len(e.steps) == 0 {
return nil
}
out := make([]tool.Step, len(e.steps))
copy(out, e.steps)
return out
}
// kindForTool maps a tool name to an open-vocabulary step kind. Unknown
// tools fall back to "tool" — never an error, just a generic step (the
// client maps unknown kinds to a default icon). Loosely tracks the
// catalog in pkg/skilltools/CLAUDE.md.
func kindForTool(name string) string {
switch name {
case "web_search", "search_reddit", "wikipedia_summary":
return "search"
case "read_page", "read_pdf", "read_reddit", "read_video", "verify_url",
"summary_summarise", "summarize", "file_get_text", "file_get_metadata",
"http_get", "http_post", "http_get_stream", "http_stream_read":
return "read"
case "code_exec", "calculate":
return "code"
case "file_save", "file_get", "file_list", "file_delete", "file_search":
return "file"
case "kv_get", "kv_set", "kv_list", "kv_delete",
"remember", "recall", "chatbot_get_memories":
return "memory"
case "query", "query_research", "deepresearch", "animate",
"agent_invoke", "agent_invoke_parallel", "agent_spawn",
"agent_spawn_parallel", "skill_invoke", "skill_invoke_parallel":
return "delegate"
case "think":
return "thinking"
default:
switch {
case strings.HasPrefix(name, "image") || strings.Contains(name, "draw"):
return "image"
default:
return "tool"
}
}
}
// summaryForStart builds the human present-tense running summary. It
// derives specifics from safe arg fields only; secret-bearing tools
// (mcp_call, email_send, http_*) are summarized without echoing args.
func summaryForStart(name string, args json.RawMessage) string {
var s string
switch name {
case "web_search":
if q := argString(args, "query", "q"); q != "" {
s = fmt.Sprintf("Searching the web for %q", q)
} else {
s = "Searching the web"
}
case "search_reddit":
if q := argString(args, "query", "q"); q != "" {
s = fmt.Sprintf("Searching Reddit for %q", q)
} else {
s = "Searching Reddit"
}
case "wikipedia_summary":
if q := argString(args, "query", "title"); q != "" {
s = fmt.Sprintf("Looking up %q on Wikipedia", q)
} else {
s = "Looking up Wikipedia"
}
case "read_page", "read_pdf", "read_reddit", "read_video", "verify_url":
if u := argString(args, "url", "post", "page"); u != "" {
s = "Reading " + hostOf(u)
} else {
s = "Reading a page"
}
case "http_get", "http_post", "http_get_stream":
// Show host only — a full URL can embed credentials/tokens.
if u := argString(args, "url"); u != "" {
s = "Fetching " + hostOf(u)
} else {
s = "Making an HTTP request"
}
case "summary_summarise", "summarize":
s = "Summarizing text"
case "translate":
if lang := argString(args, "target_lang", "target_language", "lang"); lang != "" {
s = "Translating to " + lang
} else {
s = "Translating text"
}
case "code_exec":
s = "Running code"
case "calculate":
if q := argString(args, "query", "expression", "expr"); q != "" {
s = "Calculating " + truncateStep(q, 60)
} else {
s = "Calculating"
}
case "remember":
// Never echo the stored value.
s = "Saving a memory"
case "recall", "chatbot_get_memories":
s = "Recalling memories"
case "kv_get", "kv_list":
s = "Reading saved data"
case "kv_set":
s = "Saving data"
case "kv_delete":
s = "Deleting saved data"
case "file_save":
if n := argString(args, "name", "filename"); n != "" {
s = "Saving file " + truncateStep(n, 60)
} else {
s = "Saving a file"
}
case "file_get", "file_get_text", "file_get_metadata":
s = "Reading a file"
case "file_list", "file_search":
s = "Listing files"
case "query", "query_research":
if q := argString(args, "query", "question", "prompt", "task"); q != "" {
s = "Researching " + truncateStep(q, 80)
} else {
s = "Researching"
}
case "deepresearch":
s = "Running deep research"
case "animate":
s = "Generating an animation"
case "agent_invoke", "agent_spawn":
if a := argString(args, "agent", "agent_name", "name"); a != "" {
s = "Delegating to " + a
} else {
s = "Delegating to a sub-agent"
}
case "agent_invoke_parallel", "agent_spawn_parallel":
s = "Delegating to sub-agents"
case "skill_invoke":
if sk := argString(args, "skill_name", "skill", "name"); sk != "" {
s = "Running skill " + sk
} else {
s = "Running a skill"
}
case "skill_invoke_parallel":
s = "Running skills in parallel"
case "think":
s = "Thinking"
case "mcp_call":
// Redact: MCP args frequently carry secrets. Name server/tool only.
srv, tl := argString(args, "server"), argString(args, "tool")
switch {
case srv != "" && tl != "":
s = fmt.Sprintf("Calling %s/%s", srv, tl)
case srv != "":
s = "Calling " + srv
default:
s = "Calling an MCP tool"
}
case "email_send":
// Redact recipients + body.
s = "Sending an email"
default:
s = "Using " + name
}
return truncateStep(s, stepSummaryMaxLen)
}
// summaryForEnd optionally upgrades the summary to a cheap result phrase.
// Returns "" to keep the running summary (the caller then just flips the
// status). Never returns a phrase derived from raw result bytes.
func summaryForEnd(name string, _ json.RawMessage, result string, isError bool) string {
if isError {
return ""
}
switch name {
case "web_search", "search_reddit":
if n := countResults(result); n >= 0 {
return fmt.Sprintf("Found %d result%s", n, plural(n))
}
}
return ""
}
// argString pulls the first present non-empty string field from a tool's
// raw JSON args, trying keys in order. Returns "" when none parse.
func argString(args json.RawMessage, keys ...string) string {
if len(args) == 0 {
return ""
}
var m map[string]any
if err := json.Unmarshal(args, &m); err != nil {
return ""
}
for _, k := range keys {
if v, ok := m[k]; ok {
if s, ok := v.(string); ok && strings.TrimSpace(s) != "" {
return strings.TrimSpace(s)
}
}
}
return ""
}
// countResults parses a v11-style {"results":[...]} envelope and returns
// the count, or -1 when the shape doesn't match.
func countResults(result string) int {
if strings.TrimSpace(result) == "" {
return -1
}
var env struct {
Results []json.RawMessage `json:"results"`
}
if err := json.Unmarshal([]byte(result), &env); err != nil || env.Results == nil {
return -1
}
return len(env.Results)
}
// hostOf returns the bare host (no leading www.) of a URL, or a short
// form of the raw string when it doesn't parse as a URL.
func hostOf(raw string) string {
if u, err := url.Parse(raw); err == nil && u.Host != "" {
return strings.TrimPrefix(u.Host, "www.")
}
return truncateStep(raw, 60)
}
// truncateStep rune-safely caps s to max, appending an ellipsis when cut.
func truncateStep(s string, max int) string {
if max <= 0 {
return ""
}
r := []rune(s)
if len(r) <= max {
return s
}
if max == 1 {
return string(r[:1])
}
return string(r[:max-1]) + "…"
}
func plural(n int) string {
if n == 1 {
return ""
}
return "s"
}
+84
View File
@@ -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
},
)
}
+128
View File
@@ -0,0 +1,128 @@
// Package tools — v11 cite.
//
// Anti-hallucination forcing function. The convention: agents call
// cite(claim, url) for every numbered reference in their final
// answer. The tool verifies the URL appears in the run's
// touched-URL set (populated by web_search results +
// read_page/read_pdf/read_video). If yes → write to
// skill_run_sources, return {ok: true}. If no → return
// {ok: false, reason: "url_not_in_run_history"} and DO NOT write.
//
// Skills authored without this discipline don't lose anything;
// skills WITH it produce more reliable citations. The webui
// renders the skill_run_sources rows as a Sources panel on the
// run trace page — invisible to skills that don't use cite().
package tools
import (
"context"
"encoding/json"
"fmt"
"strings"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// citeParams is the LLM-facing param struct.
type citeParams struct {
Claim string `json:"claim" description:"The claim or fact you are asserting (e.g. 'Mort was published in 1987')."`
URL string `json:"url" description:"The URL that supports the claim. MUST be a URL the agent has previously read via read_page/read_pdf/read_video or seen as a web_search result."`
}
// citeResponse is the JSON envelope returned to the agent.
//
// On success: ok=true, the skill_run_sources row was written.
// On failure: ok=false, reason=<one of the documented sentinels>.
type citeResponse struct {
OK bool `json:"ok"`
Reason string `json:"reason,omitempty"`
Claim string `json:"claim,omitempty"`
URL string `json:"url,omitempty"`
}
// NewCite constructs the v11 cite tool. cs may be nil — handler
// returns "not configured" at first call.
//
// The "anyone author / share-safe" permission shape matches every
// other v11 research-class tool. Skills that adopt cite() get the
// Sources panel automatically; skills that don't are unaffected.
func NewCite(cs CitationStorage) tool.Tool {
return tool.NewGatedTool[citeParams](
"cite",
"Record a citation: a claim + the URL that supports it. The URL MUST be one the agent has actually fetched via read_page/read_pdf/read_video or seen as a web_search result — citing a URL the agent never visited is rejected with reason 'url_not_in_run_history'. Successful citations populate the run's Sources panel.",
tool.Permission{
AuthoringRequirement: tool.RequirementAnyone,
OperatesOn: tool.ScopeGlobal,
SafeForShare: true,
Categories: []string{"citation"},
},
func(ctx context.Context, inv tool.Invocation, p citeParams) (string, error) {
if cs == nil {
return "", fmt.Errorf("cite: citation storage not configured")
}
claim := strings.TrimSpace(p.Claim)
urlStr := strings.TrimSpace(p.URL)
if claim == "" {
return marshalCite(citeResponse{
OK: false,
Reason: "claim_empty",
}), nil
}
if urlStr == "" {
return marshalCite(citeResponse{
OK: false,
Reason: "url_empty",
}), nil
}
if inv.RunID == "" {
// No run id → cite() can't verify history. Bail loud.
return marshalCite(citeResponse{
OK: false,
Reason: "no_run_context",
Claim: claim,
URL: urlStr,
}), nil
}
touched, err := cs.GetTouchedURLs(ctx, inv.RunID)
if err != nil {
return marshalCite(citeResponse{
OK: false,
Reason: fmt.Sprintf("touched_lookup_failed: %v", err),
Claim: claim,
URL: urlStr,
}), nil
}
if _, ok := touched[urlStr]; !ok {
return marshalCite(citeResponse{
OK: false,
Reason: "url_not_in_run_history",
Claim: claim,
URL: urlStr,
}), nil
}
if err := cs.RecordCitation(ctx, inv.RunID, urlStr, claim); err != nil {
return marshalCite(citeResponse{
OK: false,
Reason: fmt.Sprintf("record_failed: %v", err),
Claim: claim,
URL: urlStr,
}), nil
}
return marshalCite(citeResponse{
OK: true,
Claim: claim,
URL: urlStr,
}), nil
},
)
}
func marshalCite(r citeResponse) string {
b, err := json.Marshal(r)
if err != nil {
return fmt.Sprintf(`{"ok":false,"reason":"marshal_failed: %v"}`, err)
}
return string(b)
}
+319
View File
@@ -0,0 +1,319 @@
// Package tools — v12 classify.
//
// Classification primitive: text + categories → labels + per-category
// scores. Single-label mode (default) returns the top-1 category;
// multi-label mode returns every category whose score crosses the
// threshold.
//
// Why a dedicated tool (vs reusing extract_entities for one-of-N
// classification): classification has a typed result (labels[] +
// scores{}) that downstream agents consume programmatically. Folding
// it into extract_entities would force every author to re-spec the
// scoring schema.
//
// Score normalisation: the LLM's reply is normalised so each score
// lands in [0, 1]. The single-label result returns scores for ALL
// categories so the author can read the distribution; multi-label
// returns labels[] of categories above 0.5.
//
// Test: classify_test.go covers single-label, multi-label, score
// normalisation, > 20 categories rejected, unknown category in the
// reply silently dropped.
package tools
import (
"context"
"encoding/json"
"fmt"
"strings"
"gitea.stevedudenhoeffer.com/steve/executus/llmmeta"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// classifyMaxInputBytes is the input cap.
const classifyMaxInputBytes = 16 * 1024
// classifyMaxCategories is the hard cap on category count.
const classifyMaxCategories = 20
// classifyMultiLabelThreshold is the score threshold above which a
// category appears in the labels[] array in multi-label mode.
const classifyMultiLabelThreshold = 0.5
// classifyFallbackMaxPerRun is the per-run cap when ClassifyConfig is
// nil.
const classifyFallbackMaxPerRun = 20
// ClassifyConfig is the narrow per-deployment config surface.
type ClassifyConfig interface {
MaxPerRun(ctx context.Context) int
}
// classifyArgs is the LLM-facing param struct.
type classifyArgs struct {
Text string `json:"text" description:"The text to classify. Required. Capped at 16KB."`
Categories []string `json:"categories" description:"List of categories to score the text against. Required. Max 20."`
MultiLabel bool `json:"multi_label,omitempty" description:"When true, returns every category scoring above 0.5. Default false → single-label (top-1) result."`
}
type classifyResult struct {
Labels []string `json:"labels,omitempty"`
Scores map[string]float64 `json:"scores,omitempty"`
ModelUsed string `json:"model_used,omitempty"`
RawReply string `json:"raw_reply,omitempty"`
Error string `json:"error,omitempty"`
BudgetMsg string `json:"budget_message,omitempty"`
}
// NewClassify constructs the classify tool.
func NewClassify(helper *llmmeta.Helper, cfg ClassifyConfig, budget SearchBudget) tool.Tool {
return tool.NewGatedTool[classifyArgs](
"classify",
"Classify text into one of N categories (or multiple via multi_label=true). Returns labels[] (top-1 by default) + scores{category: 0..1}. Counts against per-run and 7-day cost budgets.",
tool.Permission{
AuthoringRequirement: tool.RequirementAnyone,
OperatesOn: tool.ScopeCaller,
SafeForShare: true,
Categories: []string{"llm-meta", "cost-bearing"},
},
func(ctx context.Context, inv tool.Invocation, args classifyArgs) (string, error) {
if helper == nil {
return "", fmt.Errorf("classify: not configured")
}
text := args.Text
if strings.TrimSpace(text) == "" {
return marshalClassifyResult(classifyResult{Error: "text is empty"}), nil
}
if len(args.Categories) == 0 {
return marshalClassifyResult(classifyResult{Error: "categories is empty"}), nil
}
if len(args.Categories) > classifyMaxCategories {
return marshalClassifyResult(classifyResult{
Error: fmt.Sprintf("too many categories (%d > %d)", len(args.Categories), classifyMaxCategories),
}), nil
}
// Trim + dedupe categories so the LLM sees a clean
// schema. Order is preserved for the prompt; the result
// map is order-agnostic.
categories := make([]string, 0, len(args.Categories))
seen := make(map[string]bool, len(args.Categories))
for _, c := range args.Categories {
c = strings.TrimSpace(c)
if c == "" || seen[c] {
continue
}
seen[c] = true
categories = append(categories, c)
}
if len(categories) == 0 {
return marshalClassifyResult(classifyResult{Error: "categories has no non-empty entries"}), nil
}
if len(text) > classifyMaxInputBytes {
text = text[:classifyMaxInputBytes]
}
// Per-run budget gate.
if budget == nil {
maxPerRun := classifyFallbackMaxPerRun
if cfg != nil {
maxPerRun = cfg.MaxPerRun(ctx)
}
budget = NewInMemorySearchBudget(map[string]int{
"classify": maxPerRun,
})
}
count, max, exceeded := budget.CheckAndIncrement(ctx, inv.RunID, "classify")
if exceeded {
return marshalClassifyResult(classifyResult{
Error: "classify_budget_exceeded",
BudgetMsg: fmt.Sprintf("per-run classify budget exceeded (%d/%d). Ask an admin to raise skills.classify.max_per_run.", count, max),
}), nil
}
systemPrompt := "You classify text into a fixed set of categories. Return ONLY JSON. Score each category in [0,1] (1 = perfect fit). Sum of all scores does NOT need to be 1 — high overlap across categories is allowed."
userPrompt := buildClassifyPrompt(text, categories, args.MultiLabel)
res, callErr := helper.Call(ctx, llmmeta.CallSpec{
Tier: "fast",
SystemPrompt: systemPrompt,
UserPrompt: userPrompt,
MaxOutputTokens: 2048,
ResponseFormat: "json",
RetryOnMalformedJSON: true,
ToolName: "classify",
RunID: inv.RunID,
SkillID: inv.SkillID,
CallerID: inv.CallerID,
})
if callErr != nil {
return "", callErr
}
if !res.Success {
kind := res.ErrorKind
if kind == "" {
kind = "llm_unavailable"
}
return marshalClassifyResult(classifyResult{Error: kind}), nil
}
if res.ErrorKind == llmmeta.ErrorKindMalformedJSON || res.Parsed == nil {
return marshalClassifyResult(classifyResult{
Error: "classification_failed",
RawReply: res.Text,
ModelUsed: res.ModelUsed,
}), nil
}
parsedMap, ok := res.Parsed.(map[string]any)
if !ok {
return marshalClassifyResult(classifyResult{
Error: "classification_failed_not_object",
RawReply: res.Text,
ModelUsed: res.ModelUsed,
}), nil
}
scores := normaliseClassifyScores(parsedMap, categories)
labels := selectClassifyLabels(scores, categories, args.MultiLabel)
return marshalClassifyResult(classifyResult{
Labels: labels,
Scores: scores,
ModelUsed: res.ModelUsed,
}), nil
},
)
}
// buildClassifyPrompt composes the user message.
func buildClassifyPrompt(text string, categories []string, multiLabel bool) string {
var sb strings.Builder
sb.WriteString("Classify the text below.\n\nCategories:\n")
for _, c := range categories {
sb.WriteString("- ")
sb.WriteString(c)
sb.WriteString("\n")
}
sb.WriteString("\nText:\n")
sb.WriteString(text)
sb.WriteString("\n\nReturn ONLY a JSON object: {\"scores\": {\"<category>\": <0..1 float>, ...}}.")
if multiLabel {
sb.WriteString(" The same text may score high in MULTIPLE categories — score each independently.")
} else {
sb.WriteString(" Score each category; the highest-scoring one will be the chosen label.")
}
return sb.String()
}
// normaliseClassifyScores extracts the scores map from the LLM's
// reply and clamps each value into [0, 1]. Categories absent from the
// reply default to 0.
//
// Why we accept either {"scores": {...}} or {...}: some models reply
// with the inner object directly, dropping the wrapping key. Both
// shapes are valid as long as the keys match the requested category
// names.
func normaliseClassifyScores(parsed map[string]any, categories []string) map[string]float64 {
scoresIn, ok := parsed["scores"].(map[string]any)
if !ok {
// Accept the bare-map shape too.
scoresIn = parsed
}
out := make(map[string]float64, len(categories))
for _, c := range categories {
v, has := scoresIn[c]
if !has {
out[c] = 0
continue
}
f, ok := coerceClassifyScore(v)
if !ok {
out[c] = 0
continue
}
// Clamp into [0, 1].
if f < 0 {
f = 0
}
if f > 1 {
f = 1
}
out[c] = f
}
return out
}
// coerceClassifyScore reads a JSON value as a float in [0, 1]. Accepts
// floats, ints, and percent-strings ("85%" → 0.85).
func coerceClassifyScore(raw any) (float64, bool) {
switch v := raw.(type) {
case float64:
return v, true
case int:
return float64(v), true
case int64:
return float64(v), true
case string:
s := strings.TrimSuffix(strings.TrimSpace(v), "%")
var f float64
if _, err := fmt.Sscanf(s, "%f", &f); err == nil {
if strings.HasSuffix(strings.TrimSpace(v), "%") {
f = f / 100.0
}
return f, true
}
}
return 0, false
}
// selectClassifyLabels picks the labels to surface. Single-label mode
// returns the highest-scoring category. Multi-label returns every
// category above the threshold (sorted by score desc for stable
// rendering).
func selectClassifyLabels(scores map[string]float64, categories []string, multiLabel bool) []string {
if multiLabel {
var labels []string
for _, c := range categories {
if scores[c] >= classifyMultiLabelThreshold {
labels = append(labels, c)
}
}
// Sort labels by score desc, then category-list order for ties.
sortClassifyLabelsByScore(labels, scores)
return labels
}
// Single-label: top-1.
bestCat := ""
bestScore := -1.0
for _, c := range categories {
if scores[c] > bestScore {
bestScore = scores[c]
bestCat = c
}
}
if bestCat == "" {
return nil
}
return []string{bestCat}
}
// sortClassifyLabelsByScore sorts labels desc by score. Stable on
// ties (preserves category-list order).
func sortClassifyLabelsByScore(labels []string, scores map[string]float64) {
for i := 1; i < len(labels); i++ {
j := i
for j > 0 && scores[labels[j]] > scores[labels[j-1]] {
labels[j], labels[j-1] = labels[j-1], labels[j]
j--
}
}
}
func marshalClassifyResult(r classifyResult) string {
b, err := json.Marshal(r)
if err != nil {
return fmt.Sprintf(`{"error":"marshal_failed: %v"}`, err)
}
return string(b)
}
+342
View File
@@ -0,0 +1,342 @@
// Package tools — v12 extract_entities.
//
// Structured-output workhorse: text + field schema → typed JSON
// object. The author specifies which fields they want and what
// types; the tool builds an appropriate prompt, asks for JSON, and
// validates + coerces the response back into the requested types.
//
// Why a structured-output tool (vs forcing the agent to write its
// own prompt): every agentic skill that needs to "pull X, Y, Z out
// of unstructured text" otherwise re-invents the same prompt-
// engineering pattern. extract_entities centralises it so authors
// just describe the schema.
//
// Type coercion: an LLM responding with "42" when an int field was
// requested is normal noise. The tool coerces strings to
// int/float/bool when possible; coercion failures land the field in
// missing_fields rather than the entities map.
//
// Test: extract_entities_test.go covers happy path, missing optional
// field, missing required field surfaces in missing_fields, malformed
// JSON retry, second-attempt failure, type coercion (string→int,
// string→bool), unknown field type rejected at args validation.
package tools
import (
"context"
"encoding/json"
"fmt"
"strconv"
"strings"
"gitea.stevedudenhoeffer.com/steve/executus/llmmeta"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// extractEntitiesMaxInputBytes is the hard input cap.
const extractEntitiesMaxInputBytes = 32 * 1024
// extractEntitiesFallbackMaxPerRun is the per-run cap when
// ExtractEntitiesConfig is nil.
const extractEntitiesFallbackMaxPerRun = 10
// ExtractEntitiesConfig is the narrow per-deployment config surface
// extract_entities reads at execute time.
type ExtractEntitiesConfig interface {
MaxPerRun(ctx context.Context) int
}
// extractField is one row in the schema the agent supplies. The four
// supported types match the JSON-shape primitives we can validate +
// coerce reliably.
//
// Why an enum-shaped Type field (vs free-form): we need to know how
// to validate the LLM's reply. Free-form ("integer", "Number",
// "boolean") would invite typos that silently miss the validation.
type extractField struct {
Name string `json:"name" description:"Field name to populate (e.g. 'author', 'year_published'). Becomes a key in the returned entities object."`
Description string `json:"description" description:"Short description of what to extract (e.g. 'the book author', 'the year the article was published'). Helps the model find the right value."`
Type string `json:"type" description:"One of: 'string', 'int', 'float', 'bool', 'list_of_strings'. Determines how the LLM's reply is validated and coerced."`
Required bool `json:"required,omitempty" description:"When true, a missing/uncoercible value lands in missing_fields rather than skipping silently."`
}
// extractEntitiesArgs is the LLM-facing param struct.
type extractEntitiesArgs struct {
Text string `json:"text" description:"The text to extract from. Required. Capped at 32KB."`
Fields []extractField `json:"fields" description:"Schema describing what to extract. Each field has name, description, type, and optional required flag."`
}
type extractEntitiesResult struct {
Entities map[string]any `json:"entities,omitempty"`
MissingFields []string `json:"missing_fields,omitempty"`
ModelUsed string `json:"model_used,omitempty"`
RawReply string `json:"raw_reply,omitempty"`
Error string `json:"error,omitempty"`
BudgetMsg string `json:"budget_message,omitempty"`
}
// validExtractTypes is the closed set of Type strings the tool
// accepts. Anything else is rejected at args validation.
var validExtractTypes = map[string]bool{
"string": true,
"int": true,
"float": true,
"bool": true,
"list_of_strings": true,
}
// NewExtractEntities constructs the extract_entities tool.
func NewExtractEntities(helper *llmmeta.Helper, cfg ExtractEntitiesConfig, budget SearchBudget) tool.Tool {
return tool.NewGatedTool[extractEntitiesArgs](
"extract_entities",
"Extract structured fields from unstructured text via a fast LLM. Caller supplies a schema (each field has name + description + type + required); tool returns an entities object with values matching the requested types. Types: string, int, float, bool, list_of_strings. Counts against per-run and 7-day cost budgets.",
tool.Permission{
AuthoringRequirement: tool.RequirementAnyone,
OperatesOn: tool.ScopeCaller,
SafeForShare: true,
Categories: []string{"llm-meta", "cost-bearing"},
},
func(ctx context.Context, inv tool.Invocation, args extractEntitiesArgs) (string, error) {
if helper == nil {
return "", fmt.Errorf("extract_entities: not configured")
}
text := args.Text
if strings.TrimSpace(text) == "" {
return marshalExtractEntities(extractEntitiesResult{Error: "text is empty"}), nil
}
if len(args.Fields) == 0 {
return marshalExtractEntities(extractEntitiesResult{Error: "fields is empty"}), nil
}
// Validate each field's Type before paying for an LLM
// call.
for _, f := range args.Fields {
if strings.TrimSpace(f.Name) == "" {
return marshalExtractEntities(extractEntitiesResult{Error: "field with empty name"}), nil
}
if !validExtractTypes[strings.ToLower(strings.TrimSpace(f.Type))] {
return marshalExtractEntities(extractEntitiesResult{
Error: fmt.Sprintf("field %q has unsupported type %q (allowed: string|int|float|bool|list_of_strings)", f.Name, f.Type),
}), nil
}
}
if len(text) > extractEntitiesMaxInputBytes {
text = text[:extractEntitiesMaxInputBytes]
}
// Per-run budget gate.
if budget == nil {
maxPerRun := extractEntitiesFallbackMaxPerRun
if cfg != nil {
maxPerRun = cfg.MaxPerRun(ctx)
}
budget = NewInMemorySearchBudget(map[string]int{
"extract_entities": maxPerRun,
})
}
count, max, exceeded := budget.CheckAndIncrement(ctx, inv.RunID, "extract_entities")
if exceeded {
return marshalExtractEntities(extractEntitiesResult{
Error: "extract_entities_budget_exceeded",
BudgetMsg: fmt.Sprintf("per-run extract_entities budget exceeded (%d/%d). Ask an admin to raise skills.extract_entities.max_per_run.", count, max),
}), nil
}
systemPrompt := "You extract structured data from unstructured text. Return ONLY valid JSON with the requested keys. If a value is not present in the text, omit the key. Do NOT invent values."
userPrompt := buildExtractPrompt(text, args.Fields)
res, callErr := helper.Call(ctx, llmmeta.CallSpec{
Tier: "fast",
SystemPrompt: systemPrompt,
UserPrompt: userPrompt,
MaxOutputTokens: 4096,
ResponseFormat: "json",
RetryOnMalformedJSON: true,
ToolName: "extract_entities",
RunID: inv.RunID,
SkillID: inv.SkillID,
CallerID: inv.CallerID,
})
if callErr != nil {
return "", callErr
}
if !res.Success {
kind := res.ErrorKind
if kind == "" {
kind = "llm_unavailable"
}
return marshalExtractEntities(extractEntitiesResult{Error: kind}), nil
}
// Second-failure malformed JSON (success=true but parsed
// is nil and ErrorKind=malformed_json). Surface the raw
// reply so the agent can salvage.
if res.ErrorKind == llmmeta.ErrorKindMalformedJSON || res.Parsed == nil {
return marshalExtractEntities(extractEntitiesResult{
Error: "extraction_failed",
RawReply: res.Text,
ModelUsed: res.ModelUsed,
}), nil
}
parsedMap, ok := res.Parsed.(map[string]any)
if !ok {
return marshalExtractEntities(extractEntitiesResult{
Error: "extraction_failed_not_object",
RawReply: res.Text,
ModelUsed: res.ModelUsed,
}), nil
}
entities, missing := coerceExtractedEntities(parsedMap, args.Fields)
return marshalExtractEntities(extractEntitiesResult{
Entities: entities,
MissingFields: missing,
ModelUsed: res.ModelUsed,
}), nil
},
)
}
// buildExtractPrompt composes the user message describing the schema
// + source text.
func buildExtractPrompt(text string, fields []extractField) string {
var sb strings.Builder
sb.WriteString("Extract the following fields from the text below. Return a JSON object with the field names as keys.\n\nFields:\n")
for _, f := range fields {
fmt.Fprintf(&sb, "- %s (%s): %s", f.Name, f.Type, f.Description)
if f.Required {
sb.WriteString(" [required]")
}
sb.WriteString("\n")
}
sb.WriteString("\nText:\n")
sb.WriteString(text)
return sb.String()
}
// coerceExtractedEntities walks the LLM's response, validating + (when
// possible) coercing each value to the requested type. Required fields
// missing or uncoercible land in missing[]; optional fields silently
// drop.
func coerceExtractedEntities(parsed map[string]any, fields []extractField) (map[string]any, []string) {
entities := make(map[string]any, len(fields))
var missing []string
for _, f := range fields {
raw, present := parsed[f.Name]
if !present || raw == nil {
if f.Required {
missing = append(missing, f.Name)
}
continue
}
value, ok := coerceFieldValue(raw, f.Type)
if !ok {
if f.Required {
missing = append(missing, f.Name)
}
continue
}
entities[f.Name] = value
}
return entities, missing
}
// coerceFieldValue attempts to convert raw to the requested type.
// Returns (value, true) on success or (nil, false) on failure.
//
// Why coerce (vs strict reject): LLMs frequently reply with strings
// that contain numbers ("42") or pseudo-booleans ("yes"). Strict
// rejection would force every author to clean the response themselves.
// Coercion is conservative — string "42" → int 42 succeeds; string
// "forty-two" → int 42 fails (the agent never asked for word-form
// parsing).
func coerceFieldValue(raw any, fieldType string) (any, bool) {
switch strings.ToLower(strings.TrimSpace(fieldType)) {
case "string":
switch v := raw.(type) {
case string:
return v, true
case float64:
return strconv.FormatFloat(v, 'f', -1, 64), true
case bool:
return strconv.FormatBool(v), true
}
return nil, false
case "int":
switch v := raw.(type) {
case float64:
// JSON numbers are float64 by default.
if v == float64(int64(v)) {
return int64(v), true
}
return nil, false
case string:
if n, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64); err == nil {
return n, true
}
// Try float-string-with-zero-fractional ("42.0").
if f, err := strconv.ParseFloat(strings.TrimSpace(v), 64); err == nil && f == float64(int64(f)) {
return int64(f), true
}
}
return nil, false
case "float":
switch v := raw.(type) {
case float64:
return v, true
case string:
if f, err := strconv.ParseFloat(strings.TrimSpace(v), 64); err == nil {
return f, true
}
}
return nil, false
case "bool":
switch v := raw.(type) {
case bool:
return v, true
case string:
s := strings.ToLower(strings.TrimSpace(v))
switch s {
case "true", "yes", "1", "y":
return true, true
case "false", "no", "0", "n":
return false, true
}
case float64:
return v != 0, true
}
return nil, false
case "list_of_strings":
switch v := raw.(type) {
case []any:
out := make([]string, 0, len(v))
for _, e := range v {
if s, ok := e.(string); ok {
out = append(out, s)
} else {
// Mixed-type lists fail the type contract.
return nil, false
}
}
return out, true
case string:
// Single-string can be lifted into a one-element list.
return []string{v}, true
}
return nil, false
}
return nil, false
}
func marshalExtractEntities(r extractEntitiesResult) string {
b, err := json.Marshal(r)
if err != nil {
return fmt.Sprintf(`{"error":"marshal_failed: %v"}`, err)
}
return string(b)
}
+49
View File
@@ -0,0 +1,49 @@
// file_storage.go declares the narrow FileStorage interface that the
// four v4 file tools (file_save, file_get, file_list, file_delete)
// need at execute time.
//
// Why a narrow interface (vs importing pkg/logic/skills directly): same
// cycle constraint as kv_storage.go — pkg/logic/skills imports
// pkg/skilltools, so we mirror the FileMeta shape here and let
// pkg/logic/mort.go adapt at wiring time.
//
// FileDomainMeta is field-for-field with skills.FileMeta; the production
// adapter is a struct copy.
package tools
import (
"context"
"errors"
"time"
)
// FileStorage is the narrow surface file tools need from the skills
// package. Production wiring (mort.go) bridges *skills.System.Storage().
// nil-safe: tools constructed against a nil FileStorage surface "not
// configured" at the first call.
type FileStorage interface {
FileSave(ctx context.Context, meta FileDomainMeta, content []byte) (string, error)
FileGet(ctx context.Context, fileID string) (*FileDomainMeta, []byte, error)
FileList(ctx context.Context, skillID, scope string) ([]FileDomainMeta, error)
FileDelete(ctx context.Context, fileID string) error
FileUsageBytes(ctx context.Context, skillID string) (int64, error)
}
// FileDomainMeta mirrors skills.FileMeta. Field-for-field; the
// production adapter is a struct copy.
type FileDomainMeta struct {
ID string // UUID, the public file_id
SkillID string
Scope string
Name string
ContentHash string // SHA256 hex
MimeType string
SizeBytes int64
CreatedAt time.Time
}
// ErrFileNotFound mirrors skills.ErrFileNotFound. The production
// adapter returns this sentinel when wrapping a skills.ErrFileNotFound;
// tools detect it with errors.Is to surface a "not_found" string to the
// LLM rather than a generic error.
var ErrFileNotFound = errors.New("file: not found")
+73
View File
@@ -0,0 +1,73 @@
package tools_test
import (
"context"
"encoding/json"
"testing"
"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
"gitea.stevedudenhoeffer.com/steve/majordomo/provider/fake"
"gitea.stevedudenhoeffer.com/steve/executus/run"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
"gitea.stevedudenhoeffer.com/steve/executus/tools"
)
// TestExecutorRunsToolUsingAgent is the end-to-end proof that a host can
// register a generic tool and the executor runs an agent that CALLS it: the
// fake model emits a `think` tool call, the executor dispatches it through the
// registered tool, then the model finalises. Exercises the full tool-dispatch
// loop + step instrumentation.
func TestExecutorRunsToolUsingAgent(t *testing.T) {
reg := tool.NewRegistry()
if err := tools.Register(reg); err != nil {
t.Fatalf("register tools: %v", err)
}
fp := fake.New("fake")
fp.Enqueue("test-model",
// Step 1: the model decides to call `think`.
fake.ReplyWith(llm.Response{
ToolCalls: []llm.ToolCall{{
ID: "call-1",
Name: "think",
Arguments: json.RawMessage(`{"thought":"plan: answer briefly"}`),
}},
}),
// Step 2: with the tool result in hand, the model finalises.
fake.Reply("all done"),
)
m, err := fp.Model("test-model")
if err != nil {
t.Fatalf("fake model: %v", err)
}
ex := run.New(run.Config{
Registry: reg,
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) {
return ctx, m, nil
},
})
res := ex.Run(context.Background(),
run.RunnableAgent{Name: "thinker", ModelTier: "test-model", LowLevelTools: []string{"think"}},
tool.Invocation{RunID: "run-tool-1", CallerID: "c"},
"do the thing")
if res.Err != nil {
t.Fatalf("run error: %v", res.Err)
}
if res.Output != "all done" {
t.Fatalf("output = %q, want %q", res.Output, "all done")
}
// The step instrumentation should have captured the think call.
var sawThink bool
for _, s := range res.Steps {
if s.Title == "think" {
sawThink = true
}
}
if !sawThink {
t.Errorf("expected a `think` step in Result.Steps, got %d steps: %+v", len(res.Steps), res.Steps)
}
}
+101
View File
@@ -0,0 +1,101 @@
package tools
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// nowParams is the LLM-facing param struct for current_time / now.
//
// Why optional `timezone`: most agent prompts know the user's local
// timezone (it's in the chatbot's system prompt) but the agent has
// no way to override on a per-call basis. An explicit arg lets a
// research skill ask "what time is it in NYC for the user reading
// this report?" without needing access to a member-config lookup
// tool.
type nowParams struct {
Timezone string `json:"timezone,omitempty" description:"Optional IANA timezone name (e.g. 'America/Chicago', 'Europe/London'). Defaults to the calling user's configured timezone, falling back to UTC."`
}
// nowResponse is the JSON envelope returned to the agent.
//
// Why a structured shape: the v1 tool returned a markdown blob.
// Agents that needed just the year had to substring-parse, which
// fails on locale variations. JSON lets the agent pick the field
// it cares about.
type nowResponse struct {
NowISO string `json:"now_iso"`
NowHuman string `json:"now_human"`
Timezone string `json:"timezone"`
Weekday string `json:"weekday"`
Year int `json:"year"`
Month int `json:"month"`
Day int `json:"day"`
Hour int `json:"hour"`
Minute int `json:"minute"`
Second int `json:"second"`
Warning string `json:"warning,omitempty"`
}
// NewNow constructs the v11 current_time / now tool. The provider
// supplies the calling member's configured timezone (per-user
// localisation). nil falls back to UTC.
//
// V11 keeps the registered tool name "now" for back-compat with the
// existing tool catalog tests AND adds the same tool surface under
// the agent-facing description "current time". The design spec
// called the tool "current_time" but the v1 registry already used
// "now" — switching the registry name would break stored skills'
// `tools` lists. Same name, expanded behaviour.
func NewNow(provider CurrentTimeProvider) tool.Tool {
return tool.NewGatedTool[nowParams](
"now",
"Return the current time. Optional 'timezone' (IANA name e.g. 'America/Chicago'); defaults to the calling user's configured timezone or UTC. Returns ISO + human-readable formats plus structured year/month/day/weekday for time-relative reasoning. Use this BEFORE assuming a year — the agent's knowledge cut-off may differ from real time.",
tool.Permission{
AuthoringRequirement: tool.RequirementAnyone,
OperatesOn: tool.ScopeGlobal,
SafeForShare: true,
Categories: []string{"utility"},
},
func(ctx context.Context, inv tool.Invocation, p nowParams) (string, error) {
tzName := strings.TrimSpace(p.Timezone)
warning := ""
if tzName == "" && provider != nil {
tzName = provider.UserTimezone(ctx, inv.CallerID)
}
if tzName == "" {
tzName = "UTC"
}
loc, err := time.LoadLocation(tzName)
if err != nil {
warning = fmt.Sprintf("unknown timezone %q; falling back to UTC", tzName)
tzName = "UTC"
loc = time.UTC
}
t := time.Now().In(loc)
out := nowResponse{
NowISO: t.Format(time.RFC3339),
NowHuman: t.Format("Monday, January 2, 2006 at 3:04 PM MST"),
Timezone: tzName,
Weekday: t.Weekday().String(),
Year: t.Year(),
Month: int(t.Month()),
Day: t.Day(),
Hour: t.Hour(),
Minute: t.Minute(),
Second: t.Second(),
Warning: warning,
}
b, mErr := json.Marshal(out)
if mErr != nil {
return "", fmt.Errorf("now: marshal: %w", mErr)
}
return string(b), nil
},
)
}
+97
View File
@@ -0,0 +1,97 @@
package tools
import (
"context"
"sync"
)
// DefaultResearchConfig returns a ResearchConfig pinned to the v11
// design defaults. Production wiring overrides via a convar-aware
// adapter; tests use the defaults directly.
func DefaultResearchConfig() ResearchConfig {
return defaultResearchConfig{}
}
type defaultResearchConfig struct{}
func (defaultResearchConfig) MaxInlineBytes(_ context.Context) int { return 12 * 1024 }
func (defaultResearchConfig) PDFMaxPages(_ context.Context) int { return 50 }
func (defaultResearchConfig) WebSearchEnabled(_ context.Context) bool { return true }
func (defaultResearchConfig) WebSearchMaxPerRun(_ context.Context) int { return 10 }
func (defaultResearchConfig) ReadPageMaxPerRun(_ context.Context) int { return 10 }
func (defaultResearchConfig) VideoMaxPerRun(_ context.Context) int { return 5 }
func (defaultResearchConfig) VerifyURLMaxPerRun(_ context.Context) int { return 20 }
func (defaultResearchConfig) ReadPDFMaxPerRun(_ context.Context) int { return 5 }
func (defaultResearchConfig) HTTPGetMaxPerRun(_ context.Context) int { return 20 }
func (defaultResearchConfig) HTTPPostMaxPerRun(_ context.Context) int { return 20 }
func (defaultResearchConfig) WebSearchAugmentThreshold(_ context.Context) int { return 5 }
// InMemorySearchBudget is the package-default SearchBudget — a
// simple per-(run,kind) counter held in a map. NOT
// production-correct because the map persists across the process
// lifetime; production wiring MUST plug a per-run reset.
//
// Why a default at all: tests want a working SearchBudget without
// rolling their own. Documenting the production-correctness gap
// here keeps the production adapter (in mort.go) honest.
type InMemorySearchBudget struct {
cap map[string]int // by kind; "" means "use Default"
mu sync.Mutex
counts map[string]int // key = runID+"|"+kind
}
// NewInMemorySearchBudget constructs a default SearchBudget. Pass a
// per-kind cap map (e.g. {"web_search": 10, "read_page": 10}); kinds
// missing from the map fall back to maxPerKindDefault.
func NewInMemorySearchBudget(caps map[string]int) *InMemorySearchBudget {
if caps == nil {
caps = map[string]int{}
}
return &InMemorySearchBudget{
cap: caps,
counts: make(map[string]int),
}
}
// CheckAndIncrement implements SearchBudget. Returns the count AFTER
// incrementing on success; the counter is NOT incremented when the
// call would exceed the cap (so a "search_budget_exceeded" rejection
// doesn't burn budget on retry).
func (b *InMemorySearchBudget) CheckAndIncrement(_ context.Context, runID, kind string) (int, int, bool) {
max := b.cap[kind]
if max <= 0 {
max = 10 // safe default
}
b.mu.Lock()
defer b.mu.Unlock()
key := runID + "|" + kind
cur := b.counts[key]
if cur >= max {
return cur, max, true
}
b.counts[key] = cur + 1
return cur + 1, max, false
}
// ResetRun is a test helper: clears the counters for a single run
// across all kinds. Production wiring uses its own per-run lifecycle
// (the executor's RunFinalizer interface).
func (b *InMemorySearchBudget) ResetRun(runID string) {
b.mu.Lock()
defer b.mu.Unlock()
prefix := runID + "|"
for k := range b.counts {
if len(k) > len(prefix) && k[:len(prefix)] == prefix {
delete(b.counts, k)
}
}
}
// StaticTimeProvider is the package-default CurrentTimeProvider —
// returns "" for every member (the tool then falls back to UTC).
// Tests that need a specific timezone wire a one-line struct.
type StaticTimeProvider struct{}
// UserTimezone implements CurrentTimeProvider with a flat fallback to "".
func (StaticTimeProvider) UserTimezone(_ context.Context, _ string) string { return "" }
+332
View File
@@ -0,0 +1,332 @@
// Package tools — research provider plumbing for v11.
//
// This file declares the narrow interfaces v11's research tools
// (web_search, read_page, read_video, read_pdf, verify_url, etc.) need
// at execute time. Production wiring lives in pkg/logic/mort.go and
// closes over the searcher chain, the extractor / chromedp client, the
// PDF extractor, and the yt-dlp wrapper.
//
// Why narrow interfaces (vs importing pkg/logic/searcher / extractor
// directly): the same cycle-break pattern used by KVStorage, FileStorage,
// HTTPConfigProvider — keeps pkg/skilltools/tools free of the wiring
// layer so tests can stub each dependency. Each provider is nil-safe:
// the tool surfaces "not configured" at first call rather than failing
// at registration.
//
// Test: each tool under pkg/skilltools/tools/ wired against these
// interfaces has its own *_test.go using the in-package fakes in
// research_providers_fakes_test.go.
package tools
import (
"context"
"errors"
"time"
)
// PageCache is the narrow surface read_page (and read_pdf) consult to
// avoid re-fetching the same URL within the cache's TTL. Production
// wiring bridges this interface to the legacy *cache.Cache held by
// pkg/logic/query.System so a `.query foo.com` and a
// `.skill query foo.com` for the same URL share one cache slot.
//
// Why a narrow interface (vs importing the cache package directly):
// same cycle-break pattern as KVStorage / FileStorage / CitationStorage
// — keeps pkg/skilltools/tools free of the wiring layer. The legacy
// cache slot key is `sha256(url)`; the production adapter is
// responsible for hashing so this interface stays clean (raw URL in/out)
// and skill-tool authors never need to know the slot shape.
//
// nil-safe: a tool constructed with a nil PageCache simply skips the
// cache layer (always treat Get as a miss; Set is a no-op).
//
// Test: tests pass a fake PageCache that records Get/Set calls and
// returns canned hits. See page_cache_test.go for the read_page hit /
// miss scenarios.
type PageCache interface {
// Get returns the cached body for urlStr and true on hit, or
// (nil, false) on miss. Implementations MUST treat any backing-
// store error as a miss (best-effort, never fail the caller).
Get(ctx context.Context, urlStr string) ([]byte, bool)
// Set writes body under the slot for urlStr with the supplied TTL.
// Implementations MUST swallow backing-store errors (best-effort
// caching is correct: a write failure should not propagate to the
// agent loop).
Set(ctx context.Context, urlStr string, body []byte, ttl time.Duration)
}
// PageCacheTTL is the default TTL applied by tools that consult a
// PageCache. Mirrors the legacy `query.pageCacheTTL` constant
// (1 hour) so a `.query`-warmed slot reads back from a `.skill query`
// (and vice versa) within the same window.
//
// Tools that want a different TTL pass an explicit value to
// PageCache.Set; this constant is the project default the v11 / v-research
// tools all use.
const PageCacheTTL = 1 * time.Hour
// PageExtractor is the narrow surface read_page needs at execute
// time. The production adapter wraps mort's existing extractor
// (Ollama web_fetch first, chromedp fallback on JS-heavy pages).
//
// nil-safe: a tool constructed with a nil PageExtractor surfaces
// "not configured" at first call.
//
// Why: read_page used to be a thin io.ReadAll over the URL — it
// missed JS rendering, didn't honour the v6 page cache, and could
// not surface the underlying provider name. v11 routes through this
// interface so the production wiring (mort.go) can plug in the
// existing query-side extractor without exposing query.Agent.
type PageExtractor interface {
// ExtractPage fetches and extracts readable text from urlStr.
// Returns the extracted body, a final URL (after any redirects
// the extractor followed), the provider name ("ollama" |
// "chromedp" | "ytdlp"), and an error.
//
// The returned body is the FULL extracted text — callers apply
// the v10 byte-vs-reference cap before surfacing to the agent.
//
// bypassCache=true skips any page cache and forces a fresh
// extraction. Default false.
ExtractPage(ctx context.Context, urlStr string, bypassCache bool) (text string, finalURL string, provider string, err error)
}
// VideoTranscriber is the narrow surface read_video needs at
// execute time. Production wiring wraps internal/ytdlp.
//
// nil-safe: tool surfaces "not configured" at first call.
//
// Why a separate interface from PageExtractor: video is a different
// shape (transcript + metadata) and a different binary (yt-dlp).
// Keeping them distinct lets tests stub each independently.
type VideoTranscriber interface {
// ExtractVideoTranscript returns the transcript text and the
// best-effort metadata (title, duration in seconds, channel).
// Implementations MUST return a non-empty transcript or an
// error — empty-transcript success is interpreted by the tool
// as a "transcript_unavailable" failure.
ExtractVideoTranscript(ctx context.Context, urlStr string) (transcript string, meta VideoMeta, err error)
}
// VideoMeta is best-effort metadata returned alongside a video
// transcript. Any field may be empty/zero if the implementation
// could not extract it.
type VideoMeta struct {
Title string
Channel string
DurationSeconds int
}
// PDFFetcher is the narrow surface read_pdf needs at execute time.
// Production wiring uses an HTTP-aware fetcher that HEAD-validates
// content-type before downloading the body.
//
// nil-safe: tool surfaces "not configured" at first call.
//
// Why: a tool that just embedded PDF extraction would couple
// fetching + parsing. Splitting the fetch (allowlist + SSRF +
// HEAD check) from the extract (page-level parsing) keeps each
// step testable and lets the same fetcher serve verify_url one
// day if we want a PDF-aware fast path.
type PDFFetcher interface {
// FetchPDF downloads the PDF at urlStr (after HEAD-validating
// content-type) and returns the raw bytes plus the final URL.
// HEAD-validation rejects a URL whose Content-Type is not a
// PDF mime AND whose path does not end in .pdf.
FetchPDF(ctx context.Context, urlStr string) (body []byte, finalURL string, err error)
}
// PDFExtractor parses PDF bytes into plain text + page count.
// Production wires internal.ExtractPDFText.
//
// Why split from PDFFetcher: tests want to vary the fetch (mock
// server returning bytes) without rebuilding the extractor.
type PDFExtractor interface {
// ExtractPDFText returns the concatenated plain-text content
// of the PDF along with the page count. The caller applies any
// per-page cap and the v10 byte-vs-reference cap on the result.
ExtractPDFText(ctx context.Context, body []byte, maxPages int) (text string, pageCount int, truncated bool, err error)
}
// HEADChecker is the narrow surface verify_url needs at execute
// time. Production wiring uses the same SSRF-pinned transport as
// http_get so the security envelope is consistent.
//
// Why a separate interface (vs reusing HTTPConfigProvider+doHTTP):
// verify_url's contract is simpler — HEAD only, no body bytes
// returned, and the agent only cares about reachable / status /
// final URL / content-type. A bespoke surface lets the production
// adapter optimise for that path (no body buffer, no body close).
type HEADChecker interface {
// HEAD performs a HEAD request against urlStr (with SSRF +
// allowlist enforcement) and returns the final URL after any
// redirects, the HTTP status code, and the Content-Type header.
// Returns reachable=false with a non-nil err for transport
// failures (DNS, TCP, allowlist rejection); reachable=true with
// any HTTP status (including 4xx/5xx) is the success shape —
// the agent decides whether the URL is "real".
HEAD(ctx context.Context, urlStr string) (finalURL string, status int, contentType string, reachable bool, err error)
}
// CitationStorage is the narrow surface cite() needs at execute
// time. Production wires *skills.System.Storage(); tests stub.
//
// nil-safe: tool surfaces "not configured" at first call.
//
// Why a narrow interface (vs importing pkg/logic/skills): same
// cycle constraint as KVStorage / FileStorage. Production adapter
// in mort.go bridges to skills.Storage's RecordCitation /
// ListCitations methods AND a separate URL-history tracker.
//
// Two responsibilities, deliberately separate:
//
// 1. RecordCitation writes a row into skill_run_sources — this is
// the user-visible citations table for the Sources panel and
// CSV export. ONLY rows the agent successfully cited via
// cite() land here.
// 2. RecordURLTouch / GetTouchedURLs maintains a per-run set of
// URLs the agent has interacted with (web_search results,
// read_page input, read_pdf input, read_video input). cite()
// reads this set to reject claims for URLs the agent never
// touched. This set lives in a different table or scope from
// the citations table — it's working state, not a record.
type CitationStorage interface {
// RecordCitation appends one (run_id, url, claim, cited_at)
// row to the citations table (skill_run_sources). cited_at is
// set by the storage layer to time.Now() when zero. The caller
// has already verified the URL is in the touched-URL set
// (via GetTouchedURLs); this method is the persistence step.
RecordCitation(ctx context.Context, runID, url, claim string) error
// RecordURLTouch records that the agent has interacted with
// `url` during `runID`. Called by web_search (per result),
// read_page, read_pdf, and read_video. Idempotent — repeat
// calls for the same (run_id, url) are no-ops at the storage
// layer.
RecordURLTouch(ctx context.Context, runID, url string) error
// GetTouchedURLs returns the set of URLs the run has
// interacted with. Used by cite() to verify that a claim's
// URL is one the agent actually visited. Empty for a fresh
// run — cite() then rejects every claim with
// "url_not_in_run_history".
GetTouchedURLs(ctx context.Context, runID string) (map[string]struct{}, error)
// ListCitations returns all citations recorded for the run, in
// insertion order. Powers the /skills/{id}/runs/{run_id}
// Sources panel.
ListCitations(ctx context.Context, runID string) ([]CitationRow, error)
}
// CitationRow mirrors the skill_run_sources row shape. Fields
// match the spec: run_id is implicit in the query, url + claim are
// what the agent submitted, cited_at is the wall-clock timestamp
// at insert.
type CitationRow struct {
URL string
Claim string
CitedAt int64 // unix-seconds; storage adapter normalises from time.Time
}
// CurrentTimeProvider exposes a "now" + per-user timezone lookup.
// Production wiring closes over the bot's member-config getter.
//
// nil-safe: a tool constructed with a nil provider falls back to
// server-time + UTC (current behaviour of NewNow before v11).
type CurrentTimeProvider interface {
// UserTimezone returns the IANA timezone name configured for
// the given Discord member ID, or "" when the member has no
// timezone configured. Empty fallback is "UTC".
UserTimezone(ctx context.Context, memberID string) string
}
// SearchBudget is the narrow surface web_search reads at execute
// time to honour skills.web_search.max_per_run.
//
// Production wiring closes over a per-run counter held by the
// executor. nil-safe: tool falls back to a built-in package
// counter (process-wide, NOT per-run) — useful for tests but NOT
// production-correct because budget bleeds across runs. The
// production adapter MUST be wired.
type SearchBudget interface {
// CheckAndIncrement returns the current count AFTER incrementing
// for the given runID, the configured max, and an error when
// the call would exceed the cap. The handler returns a clean
// "search_budget_exceeded" string on exceed (not an error so
// the agent can react).
CheckAndIncrement(ctx context.Context, runID, kind string) (count, max int, exceeded bool)
}
// ResearchConfig is the narrow surface that read_page / read_video /
// read_pdf / verify_url read at execute time for per-tool budget caps
// and inline-vs-file_id thresholds. Production wiring closes over
// the relevant convars.
//
// nil-safe: tools fall back to package defaults.
type ResearchConfig interface {
// MaxInlineBytes returns the cap above which extracted text is
// persisted as a file_id under run-scope (v10 byte-vs-reference
// principle). Default 12 KiB.
MaxInlineBytes(ctx context.Context) int
// PDFMaxPages returns the cap on pages extracted from a PDF
// before truncation. Default 50.
PDFMaxPages(ctx context.Context) int
// WebSearchEnabled is the master switch for web_search.
WebSearchEnabled(ctx context.Context) bool
// WebSearchMaxPerRun is the per-run search cap.
WebSearchMaxPerRun(ctx context.Context) int
// ReadPageMaxPerRun is the per-run page-read cap.
ReadPageMaxPerRun(ctx context.Context) int
// VideoMaxPerRun is the per-run video-read cap.
VideoMaxPerRun(ctx context.Context) int
// VerifyURLMaxPerRun is the per-run HEAD-check cap.
VerifyURLMaxPerRun(ctx context.Context) int
// ReadPDFMaxPerRun is the per-run PDF-read cap.
ReadPDFMaxPerRun(ctx context.Context) int
// HTTPGetMaxPerRun (v15.2) is the per-run http_get cap. The agent
// otherwise can retry-storm through random URLs and bloat its own
// context with each tool result. Default 20.
HTTPGetMaxPerRun(ctx context.Context) int
// HTTPPostMaxPerRun (v15.2) is the per-run http_post cap. Default 20.
HTTPPostMaxPerRun(ctx context.Context) int
// WebSearchAugmentThreshold is the minimum number of primary
// (Ollama) results required to skip the secondary (DDG/Brave)
// search. When the primary backend returns fewer than this many
// results, the augmented searcher also queries the secondary and
// merges both result sets. Default 5.
WebSearchAugmentThreshold(ctx context.Context) int
// ReplyChainDepthMax is unused here; placeholder shape for
// future per-tool caps. Kept off this interface — callers reach
// into the convar reader directly when they need it.
}
// ErrPageExtractionFailed is the sentinel returned by a PageExtractor
// when both Ollama and chromedp paths produce empty content.
var ErrPageExtractionFailed = errors.New("page extraction failed: empty content")
// ErrVideoTranscriptUnavailable is the sentinel returned by a
// VideoTranscriber when no captions / transcript could be obtained.
var ErrVideoTranscriptUnavailable = errors.New("video transcript unavailable")
// ErrPDFNotPDF is the sentinel returned by a PDFFetcher when the
// HEAD response indicates a non-PDF content-type AND the URL path
// has no .pdf extension. Surfaces a clean "url_is_not_a_pdf"
// rejection rather than a generic transport error.
var ErrPDFNotPDF = errors.New("url does not serve a PDF")
// ErrPDFEncrypted is returned by a PDFExtractor when the PDF refuses
// extraction because it is password-protected. Surfaces a clean
// "pdf_encrypted" rejection.
var ErrPDFEncrypted = errors.New("pdf is encrypted")
+113
View File
@@ -0,0 +1,113 @@
// scope_validate.go centralises the storage-scope authorisation check
// shared by every v4 KV and file tool. It enforces:
//
// - "skill" — always allowed (the skill's shared, cross-caller area).
// - "user:<callerID>" — allowed if it matches inv.CallerID (or admin).
// - "user:<other>" — allowed only for admin callers.
// - "run:<runID>" — allowed if it matches inv.RunID (or admin).
// - "run:<other>" — allowed only for admin callers.
// - "root_run:<id>" — allowed if it matches inv.RootRunID (or admin):
// the dispatch tree's SHARED scratchpad, readable
// and writable by every run under one root
// (parallel sibling workers coordinate here).
// - any other shape — rejected with a descriptive error.
//
// Why a single helper (vs inline checks in each tool): the parsing rules
// must match exactly across kv_get/set/list/delete and file_save/get/
// list/delete. Centralising them means one place to fix when the
// vocabulary evolves and one place for the test matrix.
//
// Why the isAdmin parameter: the v4 Invocation does NOT carry an
// admin flag — production tools always pass isAdmin=false. The
// parameter exists for tests (which exercise the admin paths) and for a
// future Invocation extension that adds an admin signal without
// breaking this helper's signature.
package tools
import (
"fmt"
"strings"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// ValidateScope rejects scope strings the caller is not authorised to
// access. See file-level doc for the exact ruleset.
//
// Why isAdmin is parameterised: tests pass true to verify admin paths;
// production tools currently always pass false because Invocation
// doesn't carry admin status. The gate is "you can access your own
// scope only" until a future extension threads an admin signal through
// the executor.
func ValidateScope(inv tool.Invocation, scope string, isAdmin bool) error {
if scope == "skill" {
return nil
}
if rest, ok := strings.CutPrefix(scope, "user:"); ok {
if rest == "" {
return fmt.Errorf("scope: empty user id after 'user:'")
}
if rest == inv.CallerID {
return nil
}
if isAdmin {
return nil
}
return fmt.Errorf("scope %q: cannot access another user's storage", scope)
}
if rest, ok := strings.CutPrefix(scope, "root_run:"); ok {
if rest == "" {
return fmt.Errorf("scope: empty run id after 'root_run:'")
}
// The dispatch tree's shared scratchpad. Every run in one tree
// carries the same RootRunID (stamped by both executors from the
// dispatchguard chain), so siblings spawned in parallel — even
// ephemeral workers with distinct agent IDs — validate against
// the same scope string. Storage-side, root_run scopes live in
// the shared RootRunKVPartition; this check is the isolation
// boundary between trees.
if rest == inv.RootRunID && inv.RootRunID != "" {
return nil
}
if isAdmin {
return nil
}
return fmt.Errorf("scope %q: cannot access another dispatch tree's storage", scope)
}
if rest, ok := strings.CutPrefix(scope, "run:"); ok {
if rest == "" {
return fmt.Errorf("scope: empty run id after 'run:'")
}
if rest == inv.RunID {
return nil
}
// V10: when this run is a reply continuation, the agent may
// access the PARENT run's scope. The parent's run-scope KV is
// the natural carrier for "ask user a question, save state,
// resume on reply" — without this access, every continuation
// would have to re-derive state from parent_output alone.
// Note: the parent's run-scope is subject to the v4
// auto-purge (24h after parent finished). Long-delayed replies
// will see an empty scope.
if inv.Continuation != nil && rest == inv.Continuation.ParentRunID {
return nil
}
// V14: when this run is invoked via skill_invoke /
// skill_invoke_parallel from a parent skill, the agent may
// access the PARENT run's scope. This is the natural carrier
// for the "scout fans out, parent reads consolidated state"
// pattern that deepresearch uses — research-scout writes its
// touched-URL list under run:<parent_run_id> and the parent
// reads it back during the investigate phase. Without this
// access, every parent/child handoff would have to be
// serialised through tool-result strings.
if inv.ParentRunID != "" && rest == inv.ParentRunID {
return nil
}
if isAdmin {
return nil
}
return fmt.Errorf("scope %q: cannot access another run's storage", scope)
}
return fmt.Errorf("scope %q: unknown shape; expected 'skill', 'user:<id>', 'run:<id>', or 'root_run:<id>'", scope)
}
+243
View File
@@ -0,0 +1,243 @@
// Package tools — v12 summarize.
//
// One fast-tier LLM call: text in → concise text summary out. Either
// `text` or `file_id` (mutually exclusive) supplies the source. Per-run
// budget enforced via the existing v11 SearchBudget surface (kind=
// "summarize"); per-skill cost accounting via the meta-LLM helper's
// ledger (skill_llm_meta_calls).
//
// Why a dedicated tool (vs reusing summary_summarise): summary_
// summarise wraps the URL-summary pipeline used by /summary; it's
// over-coupled to a specific extraction flow. v12's summarize is the
// "given any text, give me a summary" primitive that downstream tools
// (read_page → summarize, extract → summarize) can compose freely.
//
// File-id input path: when the caller supplies file_id, we dereference
// via FileStorage. Cross-skill check rejects stolen IDs (matching
// file_get's pattern). Scope check denies user:bob's file from alice's
// invocation.
//
// Test: summarize_test.go covers happy path (mock helper), file_id
// input, oversize input truncation, budget exceeded, focus-arg
// pass-through, cross-skill file_id rejection, and the
// missing-both-args validation.
package tools
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"gitea.stevedudenhoeffer.com/steve/executus/llmmeta"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// summarizeMaxInputBytes is the hard input cap. Inputs longer than
// this are truncated with a `truncated=true` flag in the response so
// the agent knows the summary covers a prefix.
const summarizeMaxInputBytes = 32 * 1024
// summarizeDefaultMaxWords is the default max_words when the caller
// doesn't supply one. Capped further by skills.summarize.max_words.
const summarizeDefaultMaxWords = 200
// summarizeFallbackMaxWords is the cap used when SummarizeConfig is nil.
const summarizeFallbackMaxWords = 1000
// summarizeFallbackMaxPerRun is the per-run cap used when SummarizeConfig
// is nil.
const summarizeFallbackMaxPerRun = 10
// SummarizeConfig is the narrow per-run + per-deployment config surface
// summarize reads at execute time. Production wires a closure over the
// `skills.summarize.*` convars; nil falls back to package defaults.
type SummarizeConfig interface {
MaxPerRun(ctx context.Context) int
MaxWords(ctx context.Context) int
}
// summarizeArgs is the LLM-facing param struct.
//
// Why two source fields (text + file_id) with exactly-one validation:
// the agent often produces large content via read_page / read_pdf and
// stores it as a file_id (per the v10 byte-vs-reference principle);
// forcing it to round-trip through a string would defeat the file_id
// pattern. Inline `text` is the simpler path for short snippets.
type summarizeArgs struct {
Text string `json:"text,omitempty" description:"The text to summarise. Either 'text' OR 'file_id' is required (not both). Capped at 32KB; longer inputs truncate with truncated=true in the result."`
FileID string `json:"file_id,omitempty" description:"Alternative to 'text': summarise the contents of a saved file (from read_page/read_pdf/file_save). Must belong to this skill."`
MaxWords int `json:"max_words,omitempty" description:"Maximum word count for the summary. Default 200, capped at skills.summarize.max_words (default 1000)."`
Focus string `json:"focus,omitempty" description:"Optional: what aspect to emphasise (e.g. 'security implications', 'cost analysis', 'main characters')."`
}
type summarizeResult struct {
Summary string `json:"summary"`
WordCount int `json:"word_count"`
ModelUsed string `json:"model_used"`
Truncated bool `json:"truncated,omitempty"`
BudgetMsg string `json:"budget_message,omitempty"`
Error string `json:"error,omitempty"`
}
// NewSummarize constructs the summarize tool. helper / cfg / budget /
// fileStorage may all be nil; the handler surfaces clean errors at
// first call.
func NewSummarize(helper *llmmeta.Helper, cfg SummarizeConfig, budget SearchBudget, fileStorage FileStorage) tool.Tool {
return tool.NewGatedTool[summarizeArgs](
"summarize",
"Produce a concise summary of input text using a fast LLM. Pass either 'text' or 'file_id' (one of them is required). Optional 'focus' steers the summary; 'max_words' caps length (default 200). Counts against per-run and 7-day cost budgets.",
tool.Permission{
AuthoringRequirement: tool.RequirementAnyone,
OperatesOn: tool.ScopeCaller,
SafeForShare: true,
Categories: []string{"llm-meta", "cost-bearing"},
},
func(ctx context.Context, inv tool.Invocation, args summarizeArgs) (string, error) {
if helper == nil {
return "", fmt.Errorf("summarize: not configured")
}
text, truncated, err := loadSummarizeInput(ctx, inv, args, fileStorage)
if err != nil {
return marshalSummarizeResult(summarizeResult{Error: err.Error()}), nil
}
// Per-run budget BEFORE the LLM call so a runaway loop is
// bounded.
if budget == nil {
maxPerRun := summarizeFallbackMaxPerRun
if cfg != nil {
maxPerRun = cfg.MaxPerRun(ctx)
}
budget = NewInMemorySearchBudget(map[string]int{
"summarize": maxPerRun,
})
}
count, max, exceeded := budget.CheckAndIncrement(ctx, inv.RunID, "summarize")
if exceeded {
return marshalSummarizeResult(summarizeResult{
Error: "summarize_budget_exceeded",
BudgetMsg: fmt.Sprintf("per-run summarize budget exceeded (%d/%d). Work with the summaries you already have, or ask an admin to raise skills.summarize.max_per_run.", count, max),
}), nil
}
maxWords := args.MaxWords
if maxWords <= 0 {
maxWords = summarizeDefaultMaxWords
}
cap := summarizeFallbackMaxWords
if cfg != nil {
cap = cfg.MaxWords(ctx)
}
if maxWords > cap {
maxWords = cap
}
systemPrompt := "You produce concise, accurate summaries. Honor the requested word count. Do NOT invent facts."
userPrompt := buildSummarizePrompt(text, maxWords, args.Focus)
res, callErr := helper.Call(ctx, llmmeta.CallSpec{
Tier: "fast",
SystemPrompt: systemPrompt,
UserPrompt: userPrompt,
MaxOutputTokens: maxWords * 8, // ~8 tokens per word upper bound
ResponseFormat: "text",
ToolName: "summarize",
RunID: inv.RunID,
SkillID: inv.SkillID,
CallerID: inv.CallerID,
})
if callErr != nil {
return "", callErr
}
if !res.Success || res.Text == "" {
kind := res.ErrorKind
if kind == "" {
kind = "llm_unavailable"
}
return marshalSummarizeResult(summarizeResult{Error: kind}), nil
}
summary := strings.TrimSpace(res.Text)
return marshalSummarizeResult(summarizeResult{
Summary: summary,
WordCount: countWords(summary),
ModelUsed: res.ModelUsed,
Truncated: truncated,
}), nil
},
)
}
// loadSummarizeInput resolves the input text from either args.Text or
// args.FileID. Exactly one MUST be supplied; both empty AND both
// populated are rejected.
func loadSummarizeInput(ctx context.Context, inv tool.Invocation, args summarizeArgs, fileStorage FileStorage) (string, bool, error) {
hasText := strings.TrimSpace(args.Text) != ""
hasFile := strings.TrimSpace(args.FileID) != ""
if hasText == hasFile {
// Both empty OR both populated.
if !hasText {
return "", false, fmt.Errorf("summarize: one of 'text' or 'file_id' is required")
}
return "", false, fmt.Errorf("summarize: 'text' and 'file_id' are mutually exclusive — pass one")
}
if hasText {
return capInput(args.Text)
}
if fileStorage == nil {
return "", false, fmt.Errorf("summarize: file_id input requires file storage to be configured")
}
meta, content, err := fileStorage.FileGet(ctx, args.FileID)
if err != nil {
if errors.Is(err, ErrFileNotFound) {
return "", false, fmt.Errorf("summarize: file_id not found")
}
return "", false, fmt.Errorf("summarize: file fetch: %w", err)
}
if meta.SkillID != inv.SkillID {
return "", false, fmt.Errorf("summarize: file does not belong to this skill")
}
if err := ValidateScope(inv, meta.Scope, false); err != nil {
return "", false, fmt.Errorf("summarize: %w", err)
}
return capInput(string(content))
}
// capInput truncates input to the hard byte cap, returning the
// (possibly truncated) text and a flag indicating truncation occurred.
func capInput(text string) (string, bool, error) {
if len(text) <= summarizeMaxInputBytes {
return text, false, nil
}
return text[:summarizeMaxInputBytes], true, nil
}
// buildSummarizePrompt composes the user message handed to the LLM.
func buildSummarizePrompt(text string, maxWords int, focus string) string {
var sb strings.Builder
fmt.Fprintf(&sb, "Summarise the following text in at most %d words.", maxWords)
if focus = strings.TrimSpace(focus); focus != "" {
fmt.Fprintf(&sb, " Emphasise: %s.", focus)
}
sb.WriteString("\n\n")
sb.WriteString(text)
return sb.String()
}
// countWords returns a rough word count via whitespace splitting.
// Good enough for the response's word_count column; the agent might
// see slight discrepancies vs the LLM's internal counter, which is
// acceptable.
func countWords(text string) int {
return len(strings.Fields(text))
}
// marshalSummarizeResult serialises a summarizeResult to JSON.
func marshalSummarizeResult(r summarizeResult) string {
b, err := json.Marshal(r)
if err != nil {
return fmt.Sprintf(`{"error":"marshal_failed: %v"}`, err)
}
return string(b)
}
+72
View File
@@ -0,0 +1,72 @@
// Package tools — v11 think.
//
// Pure prompt-engineering tool: the agent's "thought" is recorded
// to skill_run_logs (via the audit hook the gated wrapper applies
// transparently) but produces no side effect. The literature on
// agent design notes that giving an agent an explicit `think` tool
// keeps it on plan better than giving it nothing — without one,
// agents tend to either skip planning OR babble into the final
// output. With one, planning lands in tool calls and the final
// output stays clean.
//
// V11 deliberately rejects empty thoughts. An agent that learns
// "calling think with empty args is free" will spam it; a
// rejection forces the call to actually carry reasoning.
package tools
import (
"context"
"fmt"
"strings"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
type thinkParams struct {
Thought string `json:"thought" description:"Your reasoning. May be a plan, a working hypothesis, an analysis of a tool result, or anything else you'd note in a private scratchpad. Empty input is rejected — make this load-bearing."`
}
// thinkResponse is intentionally minimal. The agent doesn't need
// machine-readable output; the value is the audit trail + the
// implicit "now you've planned, what's next" prompting the call
// gives the agent loop.
type thinkResponse struct {
OK bool `json:"ok"`
Error string `json:"error,omitempty"`
}
// NewThink constructs the v11 think tool. No deps — the audit
// hook wrapper handles persistence transparently.
func NewThink() tool.Tool {
return tool.NewGatedTool[thinkParams](
"think",
"Record a thought / plan / working hypothesis. The thought is logged to the run trace but does NOT affect any external state. Use to slow down before a tricky tool call, sketch a multi-step plan, or summarise findings before continuing. Empty thoughts are rejected.",
tool.Permission{
AuthoringRequirement: tool.RequirementAnyone,
OperatesOn: tool.ScopeGlobal,
SafeForShare: true,
Categories: []string{"utility"},
},
func(_ context.Context, _ tool.Invocation, p thinkParams) (string, error) {
if strings.TrimSpace(p.Thought) == "" {
// Returns ok:false in a structured envelope rather
// than an error so the agent loop continues with a
// recoverable signal.
return `{"ok":false,"error":"empty_thought"}`, nil
}
// Successful think emits a flat JSON. The audit hook
// (auto-injected by NewGatedTool) writes the args + result
// pair so the trace UI shows the thought verbatim.
return `{"ok":true}`, nil
},
)
}
// Note: returning a hand-rolled JSON literal instead of a marshaller
// keeps think the cheapest possible tool — no heap allocation, no
// json.Marshal call, no goroutine-local buffer churn. The two output
// shapes are static. If a future field is added to thinkResponse,
// switch back to json.Marshal — but until then, the literal is the
// idiom that matches the tool's "do nothing" intent.
var _ = thinkResponse{} // declared so vet doesn't flag the unused struct
var _ = fmt.Errorf
+89
View File
@@ -0,0 +1,89 @@
// Package tools is executus's library of generic, host-agnostic agent tools.
//
// A host registers the tools it wants against a tool.Registry, then runs an
// agent whose RunnableAgent.LowLevelTools name them. Tools split two ways:
//
// - Always-available, zero-configuration tools register via Register (think,
// now, cite) — all nil-safe, so a light host (gadfly) calls Register and is
// immediately useful.
// - Backed tools take a nil-safe Deps describing their host backend and
// register via grouped registrars (RegisterMeta, and RegisterWeb/Store/…
// as they land). Each Deps ships sensible defaults so "some setup" is small.
//
// A host adds its own domain tools against the SAME registry.
package tools
import (
"context"
"errors"
"gitea.stevedudenhoeffer.com/steve/executus/llmmeta"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// Register adds the always-available, zero-configuration generic tools:
//
// - think — record a thought to the run trace (no external effect)
// - now — current time (UTC unless a CurrentTimeProvider is wired)
// - cite — record a source citation (inert unless a CitationStorage is wired)
//
// All are nil-safe. Returns the first registration error.
func Register(reg tool.Registry) error {
return registerAll(reg,
NewThink(),
NewNow(nil),
NewCite(nil),
)
}
// MetaDeps wires the LLM-backed meta tools (classify, extract_entities,
// summarize). Helper is required. Budget defaults to an in-memory per-run cap;
// Files is optional (summarize's file_id input is inert without it); MaxPerRun
// and MaxWords default when non-positive.
type MetaDeps struct {
Helper *llmmeta.Helper
Budget SearchBudget
Files FileStorage
MaxPerRun int // per-run cap for each meta tool; default 10
MaxWords int // summarize length cap; default 200
}
// RegisterMeta adds classify, extract_entities, and summarize. It requires a
// configured llmmeta.Helper (the fast-tier meta-LLM caller); everything else
// defaults.
func RegisterMeta(reg tool.Registry, d MetaDeps) error {
if d.Helper == nil {
return errors.New("tools: MetaDeps.Helper is required for the meta tools")
}
if d.Budget == nil {
d.Budget = NewInMemorySearchBudget(nil)
}
if d.MaxPerRun <= 0 {
d.MaxPerRun = 10
}
if d.MaxWords <= 0 {
d.MaxWords = 200
}
cfg := fixedMetaConfig{maxPerRun: d.MaxPerRun, maxWords: d.MaxWords}
return registerAll(reg,
NewClassify(d.Helper, cfg, d.Budget),
NewExtractEntities(d.Helper, cfg, d.Budget),
NewSummarize(d.Helper, cfg, d.Budget, d.Files),
)
}
func registerAll(reg tool.Registry, ts ...tool.Tool) error {
for _, t := range ts {
if err := reg.Register(t); err != nil {
return err
}
}
return nil
}
// fixedMetaConfig satisfies ClassifyConfig / ExtractEntitiesConfig /
// SummarizeConfig with static caps read from MetaDeps.
type fixedMetaConfig struct{ maxPerRun, maxWords int }
func (c fixedMetaConfig) MaxPerRun(context.Context) int { return c.maxPerRun }
func (c fixedMetaConfig) MaxWords(context.Context) int { return c.maxWords }