runtime

package
v0.0.0-...-b59f3eb Latest Latest
Warning

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

 Go to latest Published: Sep 16, 2026 License: Apache-2.0 Imports: 74 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// EnvAdmissionURL points at the platform's admission endpoint.
	// Unset → admission middleware is off; pre-#201 behavior.
	EnvAdmissionURL = "FORGE_ADMISSION_URL"

	// EnvPlatformToken is the bearer token Forge sends as
	// Authorization on every admission call. Deliberately NOT named
	// FORGE_ADMISSION_TOKEN — the platform token is reusable for
	// future Forge → platform calls (audit forwarding, telemetry
	// upload, …) without inventing one env var per surface.
	EnvPlatformToken = "FORGE_PLATFORM_TOKEN"

	// EnvOrgID + EnvWorkspaceID are the existing tenancy env vars
	// from #157 — surfaced here so the admission outbound headers
	// (`Org-Id` / `Workspace-Id`) and the inbound tenancy stamps
	// (`X-Forge-Org-ID` / `X-Forge-Workspace-ID`) read from the same
	// source. Empty values produce no header on the wire.
	EnvOrgID       = "FORGE_ORG_ID"
	EnvWorkspaceID = "FORGE_WORKSPACE_ID"
)

Environment variable names the admission middleware consumes (issue #201). Kept here as named constants so tests can reference them without string drift, and so the env surface is greppable across the codebase.

View Source 
const (
	HeaderForgeTokensIn   = "X-Forge-Tokens-In"
	HeaderForgeTokensOut  = "X-Forge-Tokens-Out"
	HeaderForgeDurationMs = "X-Forge-Duration-Ms"
	HeaderForgeModel      = "X-Forge-Model"
	HeaderForgeProvider   = "X-Forge-Provider"
)

A2A response header names for per-invocation cost telemetry. These are the inline channel for orchestrator real-time cost enforcement during parallel workflow execution — the orchestrator can ceiling-check against running totals before the next stage dispatches. They populate regardless of whether OTel tracing is enabled. See issue #87 / FWS-3.

View Source 
const (
	EnvGuardrailCaptureEvidence = "FORGE_GUARDRAIL_CAPTURE_EVIDENCE"
	EnvGuardrailRedact          = "FORGE_GUARDRAIL_REDACT"
	EnvGuardrailMaxBytes        = "FORGE_GUARDRAIL_MAX_BYTES"
)

Environment variable names mirror the existing audit/export pattern. The CLI surfaces these via run/serve flags or operators can set them directly on the agent process.

View Source 
const (
	EnvRateLimitReadRPS      = "FORGE_RATE_LIMIT_READ_RPS"
	EnvRateLimitReadBurst    = "FORGE_RATE_LIMIT_READ_BURST"
	EnvRateLimitWriteRPS     = "FORGE_RATE_LIMIT_WRITE_RPS"
	EnvRateLimitWriteBurst   = "FORGE_RATE_LIMIT_WRITE_BURST"
	EnvRateLimitCancelExempt = "FORGE_RATE_LIMIT_CANCEL_EXEMPT"
)

rateLimitEnvVars name the env var keys recognized by the runner. Exposed as constants so the CLI flag wiring + the docs reference the exact strings.

View Source 
const (
	// EnvSessionStore overrides memory.session_store: "file" | "remote".
	EnvSessionStore = "FORGE_SESSION_STORE"

	// EnvSessionStoreURL points at the platform session service. Required
	// when the backend resolves to "remote".
	EnvSessionStoreURL = "FORGE_SESSION_STORE_URL"
)

Environment variable names for the remote session store (issue #243). The remote backend deliberately reuses the admission tenancy/auth env (EnvPlatformToken / EnvOrgID / EnvWorkspaceID from admission_loader.go) so a single platform token and one set of tenancy stamps cover every Forge → platform surface.

View Source 
const DefaultGuardrailEvidenceCapBytes = 4 << 10

DefaultGuardrailEvidenceCapBytes is the per-event cap for captured evidence when GuardrailAuditConfig.MaxBytes is unset. 4 KiB matches the OTel span attribute soft cap so the same content travels through both pipelines under the same size envelope.

Variables

This section is empty.

Functions

func BuildAdmissionChecker 

func BuildAdmissionChecker(agentID string, logger coreruntime.Logger) coreruntime.AdmissionChecker

BuildAdmissionChecker resolves the admission configuration from env and returns either a PlatformAdmissionChecker (when both FORGE_ADMISSION_URL and FORGE_PLATFORM_TOKEN are set) or a NoopAdmissionChecker (when either is missing).

Partial configuration — one of the pair set, the other missing — logs a startup warning so an operator who set only the URL by mistake sees the misconfiguration in the agent log rather than silently running without admission. This mirrors the guardrails-DB startup-warn pattern from issue #166.

The agentID, orgID, and workspaceID are sourced from the agent's own configuration / env (#157) so the admission headers stay consistent with the inbound tenancy stamps on the same agent's audit events.

func BuildAgentCard 

func BuildAgentCard(workDir string, cfg *types.ForgeConfig, port int) (*a2a.AgentCard, error)

BuildAgentCard constructs an AgentCard from available sources. It first tries .forge-output/agent.json; if that doesn't exist, it falls back to the ForgeConfig.

func BuildGuardrailChecker 

func BuildGuardrailChecker(
	cfg *types.ForgeConfig,
	workDir string,
	enforce bool,
	logger coreruntime.Logger,
	auditLogger *coreruntime.AuditLogger,
	auditCfg GuardrailAuditConfig,
	tracingCfg observability.TracingConfig,
) (coreruntime.GuardrailChecker, error)

BuildGuardrailChecker creates the guardrail engine from guardrails.json (or the built-in defaults when no file is present), then applies the platform guardrails overlay (#284) so an operator can further restrict the agent's guardrails without editing the agent's file.

auditLogger and auditCfg are wired into the resulting engine so every mask/block/warn decision emits a guardrail_check event through the same sink stack the A2A handlers use. tracingCfg controls the guardrail.<gate> span instrumentation added in #161 — when CaptureContent is on, evidence is stamped on the span via the same redact-then-truncate pipeline the LLM-call content capture uses. When auditLogger is nil the engine is silent on the audit pipeline (used by tests).

A file-engine construction error logs and returns a NoopGuardrailChecker — rare, and the recovery path is well-understood.

func BuildPDPResolver 

func BuildPDPResolver(cfg *types.ForgeConfig, logger pdpLogger) *pdpResolver

BuildPDPResolver constructs the managed resolver from config + the platform identity env (reusing the admission env constants). The endpoint is already env-expanded at load (ParseForgeConfig), so startup validation saw the resolved value and an unset ${PDP_ENDPOINT} failed loud rather than reaching here empty.

func CachedGatewayToken 

func CachedGatewayToken(helperCmd string, env map[string]string) (*oauth.Token, error)

CachedGatewayToken returns the stored token for a (helper, env) WITHOUT running it. Returns (nil, nil) when none is cached. Used for token injection at client-build time and for `forge auth status`.

func ClearGatewayToken 

func ClearGatewayToken(helperCmd string, env map[string]string) error

ClearGatewayToken deletes a (helper, env) cached token (`forge auth logout`).

func DefaultPolicyScaffold 

func DefaultPolicyScaffold() *agentspec.PolicyScaffold

DefaultPolicyScaffold returns a scaffold for SkillGuardrails only. The main guardrail checks are now handled by BuildGuardrailChecker.

func DefaultStructuredGuardrails 

func DefaultStructuredGuardrails() *models.StructuredGuardrails

DefaultStructuredGuardrails returns default guardrails matching the previously built-in patterns (PII, jailbreak, secrets).

func EnsureGatewayToken 

func EnsureGatewayToken(ctx context.Context, helperCmd string, env map[string]string) (*oauth.Token, error)

EnsureGatewayToken returns a valid token for the helper, running it to (re)acquire one only when the cache is missing or within the refresh buffer of expiry. This is "login". The helper's stdout is the raw token; its expiry is read from the token's JWT exp claim (opaque/non-JWT tokens are cached with a zero expiry, i.e. re-fetched every call).

func FormatTracingStartupLine 

func FormatTracingStartupLine(cfg observability.TracingConfig) string

FormatTracingStartupLine produces a one-line human-readable summary of the resolved config for the runner's ops log at startup. Excludes Headers (may contain secrets) and ResourceAttrs (often noisy); the audit trail and the OTel collector itself are the load-bearing surfaces for those.

func GatewayCredKey 

func GatewayCredKey(helperCmd string, env map[string]string) string

GatewayCredKey is the oauth-store key for a gateway token, derived from the (helper, env) identity so the gate, overlay, and login gate all agree. It is exported so the login gate can dedup on the same identity the cache uses. A filesystem-safe form ("gateway-<hex>", no shell/path chars) keeps the plaintext fallback file name valid on every OS.

func LoadEnvFile 

func LoadEnvFile(path string) (map[string]string, error)

LoadEnvFile reads a .env file and returns key-value pairs. Missing files return an empty map and no error.

func LoadGuardrailsJSON 

func LoadGuardrailsJSON(cfg *types.ForgeConfig, workDir string) *models.StructuredGuardrails

LoadGuardrailsJSON reads guardrails.json from the project directory. Returns nil if the file does not exist.

func LoadPlatformGuardrailsOverlay 

func LoadPlatformGuardrailsOverlay() (*models.StructuredGuardrails, []string, error)

LoadPlatformGuardrailsOverlay reads the `guardrails:` overlay from every platform-policy layer (system → user → workspace, the same layers the capability policy uses) and folds them into a single most-restrictive overlay.

The overlay is authored in the YAML policy.yaml using the SAME schema as the agent's guardrails.json (guardrails.StructuredGuardrails, camelCase field names). forge-core carries it as a raw YAML subtree; here we bridge each layer YAML→JSON→typed struct and union them via MergeGuardrails.

Returns (nil, nil, nil) when no layer declares a guardrails overlay — the common case, where the agent's guardrails.json stands alone. A malformed overlay (or a policy layer that won't parse) is a hard error, matching the fail-loud posture of the capability-policy loader.

func LoadPolicyScaffold 

func LoadPolicyScaffold(workDir string) (*agentspec.PolicyScaffold, error)

LoadPolicyScaffold reads policy-scaffold.json from the output directory. Returns nil (no error) if the file does not exist. Kept for SkillGuardrails loading (separate concern from main guardrails).

func OverlaySecretsToEnv 

func OverlaySecretsToEnv(cfg *types.ForgeConfig, workDir string)

OverlaySecretsToEnv loads secrets from the config's provider chain and sets them in the OS environment so that channel adapters (which use os.Getenv) can access encrypted secrets. Only keys not already set in the env are written. workDir is the agent directory used to locate agent-local secrets.

Runs before the Runner exists (called from cmd/common.go), so it doesn't have access to the structured logger — warnings about unloadable secret files go to stderr in the same style as other early-startup messages.

func ResolveAuditPayloadCapture 

func ResolveAuditPayloadCapture(env coreruntime.AuditPayloadCapture, yaml types.AuditCaptureConfig) coreruntime.AuditPayloadCapture

ResolveAuditPayloadCapture merges a forge.yaml `audit.capture` block on top of an env-derived `AuditPayloadCapture` and returns the effective config the runner should hand to registerAuditHooks.

Precedence (high → low):

  1. forge.yaml `audit.capture.*` — any non-nil bool / non-zero int wins over the env layer below.
  2. Env vars `FORGE_AUDIT_CAPTURE_*` — already baked into the env parameter via AuditPayloadCaptureFromEnv.
  3. Zero / safe defaults — every capture flag false, Redact true.

Why `*bool` in the yaml layer: an operator who writes `tool_args: false` in forge.yaml is making an explicit choice, not "fall through to env." Booleans need a nullable representation to preserve that distinction; ints get the same treatment via 0=unset for MaxBytes.

MaxBytes when set populates all four CapXxxBytes fields uniformly (matching the env-layer single-knob semantic). Operators who need different caps per field embed Forge as a library and set AuditPayloadCapture programmatically; per-field env / yaml knobs would inflate the operator surface without clear demand. See issue #163.

func ResolveRateLimit 

func ResolveRateLimit(cfg *types.ForgeConfig, override *RateLimitOverride) *server.RateLimitConfig

ResolveRateLimit merges, in precedence order:

  1. CLI override (the *RateLimitOverride passed in by the cmd layer)
  2. FORGE_RATE_LIMIT_* env vars
  3. cfg.Server.RateLimit from forge.yaml
  4. server.defaultRateLimitConfig (the bumped FWS-10 defaults)

Returned pointer is suitable for ServerConfig.RateLimit. nil means "no overrides anywhere; let the server install its own defaults" — the common case for a forge.yaml that doesn't mention rate limits.

See issue #110 / FWS-10.

func ResolveTracingConfig 

func ResolveTracingConfig(
	yamlCfg types.TracingYAML,
	flags TracingFlags,
	agentID, agentVersion, runtimeVersion string,
) observability.TracingConfig

ResolveTracingConfig folds three sources into one observability.TracingConfig that observability.NewTracerProvider can consume. Precedence, lowest → highest:

  1. Built-in defaults (DefaultProtocol / DefaultSampler / ...)
  2. `observability.tracing:` in forge.yaml (types.TracingYAML)
  3. OTEL_* environment variables (the standard ones every OTel SDK reads — operators arrive with these already in muscle memory)
  4. CLI flags (--otel-*; a deploy-time override that wins over yaml and env)

Two derived fields the operator does not set themselves:

  • ServiceName falls back to agentID when nothing else supplies one.
  • ServiceVersion is copied from agentVersion (the agent's forge.yaml `version:`).
  • RuntimeVersion is the Forge cli's own build version, surfaced as the `forge.runtime.version` resource attribute.

Pure function — no side effects, no logging. Caller decides what to do with the result (typically: feed it to observability.NewTracerProvider, which returns ErrDisabled when Enabled is false or Endpoint is empty).

func SkillScriptCandidatePaths 

func SkillScriptCandidatePaths(skillDirName, toolName string) []string

SkillScriptCandidatePaths returns the container-relative script paths a `## Tool: toolName` in skill directory skillDirName can bind to, in resolution priority order: skill-local scripts/ before the shared skills/scripts/, shell before python before node, hyphenated name before underscore. It returns nil when the tool name would escape the tree. Exported so `forge skills validate` checks the tool→script binding against the exact same candidate set the runtime resolves against — the two can never drift.

func SkillScriptExtensions 

func SkillScriptExtensions() []string

SkillScriptExtensions returns the recognized skill-script file extensions (each with a leading dot), for tooling that enumerates a scripts/ directory — e.g. `forge skills validate`'s orphan-script check.

func VerifyBuildOutput 

func VerifyBuildOutput(outputDir string) error

VerifyBuildOutput verifies the integrity of build output files against checksums.json. Returns nil if checksums.json is not found (verification is optional). Returns an error if any file's checksum doesn't match or if the signature is invalid.

func WriteStepUpChallengeOnError 

func WriteStepUpChallengeOnError(w http.ResponseWriter, err error) bool

WriteStepUpChallengeOnError inspects err for a *stepup.RequiredError and, if present, writes an RFC 9470 step-up challenge to w: 

HTTP/1.1 401 Unauthorized
WWW-Authenticate: Bearer error="step_up_required",
                         acr_values="<RequiredAcr>"

Returns true when it handled the response — the caller MUST NOT write a second response body. Returns false when err is not a step-up error, letting the caller fall through to its default error handler.

Split out so the three tasks/send handler variants (JSON-RPC, REST, SSE) all get the same challenge format via one code path.

Types

type AuthorizeURLProvider 

type AuthorizeURLProvider func(ctx context.Context, subject, server string) (string, error)

AuthorizeURLProvider supplies the consent link the deliverer presents to the user. Standalone builds it here (buildStandaloneConsentLink); a managed platform can supply its own pre-built URL via SetAuthorizeURLProvider (the seam #343's Slack delivery consumes so the same delivery code serves both modes). Returns the link to open, or an error the deliverer surfaces.

type CallbackCompleter 

type CallbackCompleter func(ctx context.Context, subject, server, code, verifier string) error

CallbackCompleter exchanges an OAuth authorization code (with the PKCE verifier bound to the state) for a token and stores it for {subject, server}, so the resumed call finds a grant. The standalone resolver provides one; when nil the loopback callback is NOT registered (managed mode hosts its own callback and never hands Forge a code). ctx is the request context so the token exchange inherits a finite deadline.

type ChecksumsFile 

type ChecksumsFile struct {
	Version   string            `json:"version"`
	Checksums map[string]string `json:"checksums"`
	Timestamp string            `json:"timestamp"`
	Signature string            `json:"signature,omitempty"`
	KeyID     string            `json:"key_id,omitempty"`
}

ChecksumsFile mirrors the JSON structure written by the signing stage.

type ConsentDeliverer 

type ConsentDeliverer func(ctx context.Context, subject, server, taskID string, deadline time.Time) error

ConsentDeliverer delivers an MCP auth-required consent prompt to the requesting user (e.g. a Slack DM with a "Connect Atlassian" link, or an A2A `auth-required` artifact). It is the auth-gate analog of DeferralNotifier.

Mode split (design-tool-registry.md §18.4): in MANAGED mode the platform owns delivery + the consent callback + token custody, so Forge is handed a deliverer that hands off to the platform. In STANDALONE mode the default is nil (no delivery yet) until the loopback resolver lands (#330 inc 4); the gate still parks and the resume endpoint still works, so an operator or the platform can drive consent out-of-band.

Best-effort: a delivery error is logged, never fatal — the parked call still resumes when a grant arrives via the resume endpoint, and blocking on a channel outage would be strictly worse.

type DecisionResolver 

type DecisionResolver interface {
	Resolve(ctx context.Context, hctx *coreruntime.HookContext) Verdict
}

DecisionResolver reaches an authorization verdict for a proposed tool call. It MUST NEVER fail open — every error is a Deny Verdict, never allow-on-error and never a zero-value Verdict (see the contract note above). A caching resolver can decorate an implementation later without touching the hook or loop.go.

type DeferralNotifier 

type DeferralNotifier func(ctx context.Context, to, taskID, tool, approverContext string, timeout time.Duration) error

DeferralNotifier is called when a tool call is deferred for human approval (R4c #211) to deliver an interactive approval request to a channel (#310). `to` is the tool's `security.defer.tools.<tool>.to` value (e.g. "channel:slack:#oncall"). Optional — a nil notifier means no channel delivery; the approver can still POST /tasks/{id}/decisions directly. A delivery error is logged, never fatal (a Slack outage must not auto-deny).

type FileWatcher 

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

FileWatcher polls the filesystem for changes and invokes a callback.

func NewFileWatcher 

func NewFileWatcher(dir string, onChange func(), logger coreruntime.Logger) *FileWatcher

NewFileWatcher creates a watcher that polls dir every 2s for changes in watched file types. onChange is called (debounced) when changes are detected.

func (*FileWatcher) Watch 

func (w *FileWatcher) Watch(ctx context.Context)

Watch starts polling until ctx is cancelled. It blocks.

type GuardrailAuditConfig 

type GuardrailAuditConfig struct {
	// CaptureEvidence includes the raw triggering content in the
	// emitted guardrail_check event's `fields.evidence`. OFF by default.
	CaptureEvidence bool

	// Redact runs a known-secret regex pass on the captured evidence
	// before truncation. ON by default. Disable only when consuming
	// in an environment that has its own scrubbing layer (e.g. a
	// platform-side SIEM normalizer).
	Redact bool

	// MaxBytes is the soft cap on the captured evidence string. Zero
	// uses DefaultGuardrailEvidenceCapBytes (4 KiB).
	MaxBytes int
}

GuardrailAuditConfig controls how the LibraryGuardrailEngine emits guardrail_check audit events. The default zero value preserves the pre-#155 metadata-only posture: an emitted event carries direction, decision, guardrail type, and violation count, but never the raw content that triggered the rule.

Operators who need the offending text (to tune patterns, debug false positives, or satisfy compliance evidence requirements) opt in by flipping CaptureEvidence to true. The Redact knob is on by default and runs an obvious-secret scrub even on the captured evidence, so a leaked API key in a prompt does not get re-published into the audit stream verbatim. MaxBytes bounds the captured substring per event; zero falls back to DefaultGuardrailEvidenceCapBytes.

Same posture as the #130 OTel content-capture work: default off, opt-in per-deployment, redact-then-truncate when on.

func GuardrailAuditConfigFromEnv 

func GuardrailAuditConfigFromEnv() GuardrailAuditConfig

GuardrailAuditConfigFromEnv reads the env vars and returns a populated config. Redact defaults to true so flipping CaptureEvidence on without touching Redact preserves the safer posture.

type GuardrailTightening 

type GuardrailTightening struct {
	// Field is the dotted path into StructuredGuardrails, e.g.
	// "security.commandInjection.action" or "gateConfig.outputGate".
	Field string
	// Change is a short human-readable before→after, e.g. "warn -> block"
	// or "enabled" or "+2 rules".
	Change string
}

GuardrailTightening records one place where the platform overlay made an agent's guardrails STRICTER. Emitted for audit so an operator can see exactly what a layer changed (mirrors the violation attribution in forge-core/security/platform_policy_enforce.go).

func MergeGuardrails 

func MergeGuardrails(agent, platform *models.StructuredGuardrails) (*models.StructuredGuardrails, []GuardrailTightening)

MergeGuardrails returns the agent's guardrails tightened by the platform overlay — a one-way ratchet: the platform can force detections/gates ON, raise actions, lower thresholds, and union rule/denylist/blocked-skill sets, but can NEVER loosen anything the agent declared. An absent platform section leaves the agent's setting untouched.

The returned value is a deep copy — neither input is mutated. The second return is the list of tightenings the platform applied, for audit.

type K8sBackendConfig 

type K8sBackendConfig struct {
	// ServiceURL is the in-cluster URL CronJob trigger pods POST to.
	// When empty, the constructor derives the standard in-cluster
	// Service DNS: http://<agent_id>.<namespace>.svc:<port>/ . This
	// matches the value the build-time schedule-manifest stage stamps
	// into generated CronJob YAML (see forge-cli/build/schedule_manifest_stage.go).
	// Operators set this explicitly when the agent listens on a
	// non-standard port or sits behind an Ingress / Gateway.
	ServiceURL string
	// Port is the port the agent's A2A server listens on; combined
	// with agent_id and namespace to derive ServiceURL when unset.
	// Defaults to 8080 when zero (matches the runner's listen-port
	// default in forge-cli/runtime/runner.go).
	Port int
	// AuthSecretName is the K8s Secret containing the internal bearer
	// token CronJobs mount. Defaults to "<agent_id>-internal-token"
	// when empty (matches `forge auth secret-yaml`).
	AuthSecretName string
	// TriggerImage is the container image the CronJob's trigger pod
	// runs. Defaults to scheduler.DefaultTriggerImage when empty.
	TriggerImage string
	// AllowDynamic gates whether Set / Delete calls (from the LLM
	// `schedule_set` / `schedule_delete` builtin tools) can create or
	// remove CronJobs at runtime. Default false — Set returns a
	// clear error explaining the rationale. Sync (declarative) is
	// always allowed regardless of this flag.
	AllowDynamic bool
}

K8sBackendConfig carries the runtime tuning the KubernetesBackend needs above and beyond the CronJob manifest defaults. Sourced from forge.yaml's `scheduler.kubernetes` block + the resolved agent_id + in-cluster service URL.

type KubernetesBackend 

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

KubernetesBackend implements scheduler.Backend by delegating persistence + timing to the cluster's CronJob controller. See docs/deployment/scheduler-kubernetes.md and issue #162.

func NewKubernetesBackend 

func NewKubernetesBackend(agentID, namespace string, cfg K8sBackendConfig, logger coreruntime.Logger) (*KubernetesBackend, error)

NewKubernetesBackend builds a backend wired to an in-cluster kubernetes.Interface. The namespace defaults to the pod's own namespace, read from the standard projected /var/run/secrets/kubernetes.io/serviceaccount/namespace file when the caller passes "". Returns a typed error when not running in-cluster (the in-cluster config probe fails) — the runner surfaces this as a startup abort when `scheduler.backend: kubernetes` was explicitly requested.

func NewKubernetesBackendWithClient 

func NewKubernetesBackendWithClient(client kubernetes.Interface, agentID, namespace string, cfg K8sBackendConfig, logger coreruntime.Logger) *KubernetesBackend

NewKubernetesBackendWithClient is the testing seam: callers (unit tests against `fake.Clientset`) pass an explicit kubernetes.Interface instead of probing the in-cluster config. Production code uses NewKubernetesBackend.

func (*KubernetesBackend) Delete 

func (b *KubernetesBackend) Delete(ctx context.Context, id string) error

Delete removes a schedule by ID. Gated by AllowDynamic when the target CronJob is LLM-sourced; declarative (yaml-sourced) CronJobs are only removed by Sync's reconciliation path, not by direct Delete calls.

func (*KubernetesBackend) Get 

func (b *KubernetesBackend) Get(ctx context.Context, id string) (*scheduler.Schedule, error)

Get returns a single schedule by ID. Returns (nil, nil) when the matching CronJob is absent — same contract as ScheduleStore.Get.

func (*KubernetesBackend) History 

func (b *KubernetesBackend) History(_ context.Context, _ string, _ int) ([]scheduler.HistoryEntry, error)

History returns an empty list with a one-time warning. K8s CronJob status carries LastScheduleTime + a small Job-history window, but the canonical source of truth in K8s mode is the audit stream (`schedule_fire` / `schedule_complete` events). The schedule_history builtin tool reads from there.

func (*KubernetesBackend) List 

func (b *KubernetesBackend) List(ctx context.Context) ([]scheduler.Schedule, error)

List returns every Forge-owned CronJob in the namespace.

func (*KubernetesBackend) Reload 

func (b *KubernetesBackend) Reload(_ context.Context)

Reload is a no-op — every Backend method hits the API directly, no cached state to refresh.

func (*KubernetesBackend) Set 

func (b *KubernetesBackend) Set(ctx context.Context, s scheduler.Schedule) error

Set creates or updates a single schedule. Gated by AllowDynamic when the source is not "yaml": the LLM-driven schedule_set tool reaches this path via the schedule_set builtin tool the runner registers. Returns a clear error when dynamic creation is disabled.

Declarative sources (Sync with source=yaml) bypass AllowDynamic.

func (*KubernetesBackend) Start 

func (b *KubernetesBackend) Start(_ context.Context)

Start is a no-op — the cluster's CronJob controller owns timing. No goroutines, no ticker.

func (*KubernetesBackend) Stop 

func (b *KubernetesBackend) Stop()

Stop is a no-op for the same reason.

func (*KubernetesBackend) Sync 

func (b *KubernetesBackend) Sync(ctx context.Context, declared []scheduler.Schedule) error

Sync reconciles cluster CronJobs against the declared yaml entries.

  • For each declared entry: create the CronJob if absent, patch when the spec drifted, leave alone when in sync.
  • For each EXISTING yaml-sourced CronJob NOT in declared: delete.
  • LLM-sourced CronJobs (label forge.schedule.source=llm) are left alone regardless of the declared list — the LLM owns them via Set / Delete.

Mirrors the FileBackend.Sync rule from forge-core/scheduler/backend.go.

type LibraryGuardrailEngine 

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

LibraryGuardrailEngine implements coreruntime.GuardrailChecker using the github.com/initializ/guardrails library. Config is a StructuredGuardrails loaded from guardrails.json (optionally tightened by the platform guardrails overlay — see #284).

On every mask / block / warn decision the engine emits a guardrail_check audit event through auditLogger (when wired). The fields.gate value carries the library gate type (input / context / tool_call / output / stream) — see issue #159 for the unified gate model. The auditCfg knob controls whether the offending content is captured as evidence (off by default — issue #155).

func NewFileGuardrailEngine 

func NewFileGuardrailEngine(sg *models.StructuredGuardrails, enforce bool, logger coreruntime.Logger) (*LibraryGuardrailEngine, error)

NewFileGuardrailEngine creates a guardrail engine backed by a local StructuredGuardrails config (loaded from guardrails.json).

func (*LibraryGuardrailEngine) CheckContext 

func (e *LibraryGuardrailEngine) CheckContext(ctx context.Context, content string) (string, error)

CheckContext validates retrieved context (system messages, RAG chunks, memory recall content) via ContextGate before it is injected into the LLM prompt. Returns the (possibly masked) content. Wired from the BeforeLLMCall hook in the runner.

func (*LibraryGuardrailEngine) CheckInbound 

func (e *LibraryGuardrailEngine) CheckInbound(ctx context.Context, msg *a2a.Message) (coreruntime.PolicyResult, error)

CheckInbound validates an inbound (user) message via InputGate. The returned PolicyResult carries the engine's Allow/Deny/Modify decision (see #209). On Modify, msg.Parts is mutated in place so downstream reads see the redacted text.

func (*LibraryGuardrailEngine) CheckOutbound 

func (e *LibraryGuardrailEngine) CheckOutbound(ctx context.Context, msg *a2a.Message) (coreruntime.PolicyResult, error)

CheckOutbound validates an outbound (agent) message via OutputGate. Masked content is applied in-place; blocked content returns an error only in enforce mode. One guardrail.output span per text part — the trace tree mirrors the part-level iteration.

The aggregated PolicyResult follows the MOST-RESTRICTIVE part's outcome per the PolicyDecision ordering (Allow < Modify < StepUp < Defer < Deny). The strict `>` comparison keeps the FIRST part at each severity level — subsequent equal-severity parts do not overwrite Modified. In practice callers today only inspect the mutated msg.Parts (each part carries its own redaction) — the aggregate.Modified string is scaffolding for future R4b/R4c callers that need a single-string projection.

func (*LibraryGuardrailEngine) CheckStream 

func (e *LibraryGuardrailEngine) CheckStream(ctx context.Context, chunk string) (string, error)

CheckStream validates a single chunk from a streaming LLM call via StreamGate. Returns the (possibly masked) chunk. Not auto-wired because Forge's current Execute loop does not call provider streaming (ExecuteStream buffers a single non-streaming response). Exposed for callers that consume llm.Client.ChatStream directly and for future loop work that adds a real per-chunk seam.

func (*LibraryGuardrailEngine) CheckToolCall 

func (e *LibraryGuardrailEngine) CheckToolCall(ctx context.Context, toolName, args string) (string, error)

CheckToolCall validates the arguments the agent is about to pass to a tool via ToolCallGate. Returns the (possibly masked) args. Wired from the BeforeToolExec hook in the runner.

func (*LibraryGuardrailEngine) CheckToolOutput 

func (e *LibraryGuardrailEngine) CheckToolOutput(ctx context.Context, toolName, text string) (string, error)

CheckToolOutput scans tool output text via OutputGate. Returns the (possibly masked) text and any blocking error. The emitted event carries fields.tool so SIEM consumers can distinguish output-gate fires on tool results from output-gate fires on the model's reply to the user.

func (*LibraryGuardrailEngine) WithAuditLogger 

func (e *LibraryGuardrailEngine) WithAuditLogger(al *coreruntime.AuditLogger, cfg GuardrailAuditConfig) *LibraryGuardrailEngine

WithAuditLogger wires an AuditLogger and capture config so the engine can emit guardrail_check events on every mask/block/warn decision. Returns the receiver for fluent construction. When auditLogger is nil the engine is silent on the audit pipeline (legacy behavior — only the ops logger sees the redaction line). Callers in the runner pass the same AuditLogger they hand to the A2A handlers so events share the configured sink stack.

func (*LibraryGuardrailEngine) WithTracing 

func (e *LibraryGuardrailEngine) WithTracing(cfg observability.TracingConfig) *LibraryGuardrailEngine

WithTracing wires the runtime's TracingConfig so the engine can stamp forge.guardrail.evidence with the redact-then-truncate pipeline when CaptureContent is enabled. Same posture as the LLM call content capture from issue #130 — default off, opt-in per deployment, redact on by default when on. Returns the receiver for fluent construction.

The guardrail.<gate> spans are opened unconditionally (the noop tracer's overhead is near-zero); CaptureContent only gates whether the evidence attribute is set. See issue #161.

type LocalSession 

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

LocalSession is an in-process agent runtime for `forge try` (issue #350). It assembles the SAME coreruntime.LLMExecutor that `forge run` uses — the built-in tool registry, egress enforcement (in-process client + subprocess proxy), audit + progress hooks, and provider client — but WITHOUT an HTTP server, scheduler, MCP, admission, auth, or long-term memory. There is no second executor: this is a trimmed bootstrap around the shared sub-builders.

Turns run one at a time via RunTurn. Conversation history is kept in memory and never persisted (the executor Store is nil), so nothing touches disk for the ephemeral run.

func NewLocalSession 

func NewLocalSession(ctx context.Context, opts LocalSessionOptions) (*LocalSession, error)

NewLocalSession builds the in-process executor for the demo agent. The order mirrors Run(): resolve env → egress → tools → model → hooks → executor.

func (*LocalSession) AuditLogger 

func (s *LocalSession) AuditLogger() *coreruntime.AuditLogger

AuditLogger exposes the session's audit logger so the visible-loop renderer (Phase 4) can attach itself as an additional sink.

func (*LocalSession) Close 

func (s *LocalSession) Close() error

Close stops the egress proxy and releases executor resources.

func (*LocalSession) RunTurn 

func (s *LocalSession) RunTurn(ctx context.Context, prompt string, progress coreruntime.ProgressEmitter) (string, error)

RunTurn runs exactly one agent turn: the prompt plus the accumulated history, through the shared executor. It installs the egress-enforced client and the optional progress emitter on the context, appends the user + agent messages to history, and returns the agent's text reply.

type LocalSessionOptions 

type LocalSessionOptions struct {
	Config       *types.ForgeConfig
	WorkDir      string
	EnvOverrides map[string]string // credential env from the paste-key picker (else nil)
	Verbose      bool
}

LocalSessionOptions configure an in-process `forge try` session.

type MemoryScheduleStore 

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

MemoryScheduleStore implements scheduler.ScheduleStore backed by a markdown file.

func NewMemoryScheduleStore 

func NewMemoryScheduleStore(path string) *MemoryScheduleStore

NewMemoryScheduleStore creates a store at the given file path.

func (*MemoryScheduleStore) Delete 

func (s *MemoryScheduleStore) Delete(_ context.Context, id string) error

func (*MemoryScheduleStore) Get 

func (s *MemoryScheduleStore) Get(_ context.Context, id string) (*scheduler.Schedule, error)

func (*MemoryScheduleStore) History 

func (s *MemoryScheduleStore) History(_ context.Context, scheduleID string, limit int) ([]scheduler.HistoryEntry, error)

func (*MemoryScheduleStore) List 

func (s *MemoryScheduleStore) List(_ context.Context) ([]scheduler.Schedule, error)

func (*MemoryScheduleStore) RecordRun 

func (s *MemoryScheduleStore) RecordRun(_ context.Context, entry scheduler.HistoryEntry) error

func (*MemoryScheduleStore) Set 

func (s *MemoryScheduleStore) Set(_ context.Context, sched scheduler.Schedule) error

type MockExecutor 

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

MockExecutor implements AgentExecutor with canned responses. It produces the same output format as MockRuntime for backward compatibility.

func NewMockExecutor 

func NewMockExecutor(tools []agentspec.ToolSpec) *MockExecutor

NewMockExecutor creates a MockExecutor with the given tool specs.

func (*MockExecutor) Close 

func (m *MockExecutor) Close() error

Close is a no-op for MockExecutor.

func (*MockExecutor) Execute 

func (m *MockExecutor) Execute(ctx context.Context, task *a2a.Task, msg *a2a.Message) (*a2a.Message, error)

Execute returns a message with mock text content.

func (*MockExecutor) ExecuteStream 

func (m *MockExecutor) ExecuteStream(ctx context.Context, task *a2a.Task, msg *a2a.Message) (<-chan *a2a.Message, error)

ExecuteStream wraps Execute as a single-item channel.

type MockRuntime 

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

MockRuntime implements AgentRuntime without a real subprocess. It returns canned responses based on the agent's tool specs. Useful for testing the A2A protocol layer without needing Python or other frameworks installed.

func NewMockRuntime 

func NewMockRuntime(tools []agentspec.ToolSpec) *MockRuntime

NewMockRuntime creates a MockRuntime with the given tool specs.

func (*MockRuntime) Healthy 

func (m *MockRuntime) Healthy(ctx context.Context) bool

func (*MockRuntime) Invoke 

func (m *MockRuntime) Invoke(ctx context.Context, taskID string, msg *a2a.Message) (*a2a.Task, error)

Invoke returns a completed task with mock text content.

func (*MockRuntime) Restart 

func (m *MockRuntime) Restart(ctx context.Context) error

func (*MockRuntime) Start 

func (m *MockRuntime) Start(ctx context.Context) error

func (*MockRuntime) Stop 

func (m *MockRuntime) Stop() error

func (*MockRuntime) Stream 

func (m *MockRuntime) Stream(ctx context.Context, taskID string, msg *a2a.Message) (<-chan *a2a.Task, error)

Stream wraps Invoke as a single-item channel.

type PlatformAdmissionChecker 

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

PlatformAdmissionChecker calls a platform-side admission endpoint per inbound request (cached for admissionCacheTTL) to decide whether the agent should admit new work. Built for issue #201.

The checker is hard-coded to fail open: any failure path — network error, timeout, 4xx, 5xx, parse error — turns into a logged warning plus an `Allowed: true, Fallback: true` Decision that gets cached for the TTL. Operators who need hard enforcement on platform outage handle it at a different layer (ingress, NetworkPolicy). The default posture trades hard enforcement for availability — the cascade of "platform is degraded → every agent stops serving" is a worse production failure than "platform is degraded → quotas leak a bit for the duration."

The cache is keyed on a single string: the agent process is asking-about-itself, so there's one decision in flight per process at any moment. agentID / orgID / workspaceID are read at startup from env (#157) and don't change at runtime.

func NewPlatformAdmissionChecker 

func NewPlatformAdmissionChecker(
	url, agentID, orgID, workspaceID, platformToken string,
	logger coreruntime.Logger,
) *PlatformAdmissionChecker

NewPlatformAdmissionChecker constructs a checker against the given endpoint with baked timeout + caching. Returns the checker ready to serve — no health-check at construction time. The first Admit call hits the platform; if it fails, the fallback-admit posture kicks in and the warn log surfaces in the operator's pipeline.

agentID is required; the platform's URL routes on it. orgID and workspaceID are optional (empty → header omitted on the wire so the platform parser distinguishes "unset" from "empty string").

func (*PlatformAdmissionChecker) Admit 

func (c *PlatformAdmissionChecker) Admit(ctx context.Context) coreruntime.Decision

Admit returns a Decision for the current request. Wraps the platform call in an admission.check OTel span; the underlying http.client call nests beneath it via the default transport (the runner installs otelhttp-wrapped transports on the egress client, not on this internal admission client — admission calls aren't counted toward LLM-provider egress).

Cache semantics: hit within TTL → return cached Decision unchanged (Cached=true is overlaid for span / audit visibility, but the underlying decision is byte-identical). Miss → synchronous call, cache result for TTL, return it. Failure → log warn, cache an admit with Fallback=true for TTL.

type RateLimitOverride 

type RateLimitOverride struct {
	ReadRPS      *float64
	ReadBurst    *int
	WriteRPS     *float64
	WriteBurst   *int
	CancelExempt *bool
}

RateLimitOverride carries values from any one configuration layer (CLI flags, env vars, or forge.yaml). Pointer-typed fields so the resolver can distinguish "unset → fall through to next layer" from "explicitly set to zero / false" (which must win over a non-zero default). The cmd layer populates one of these per layer, the resolver merges them in precedence order.

type Runner 

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

Runner orchestrates the local A2A development server.

func NewRunner 

func NewRunner(cfg RunnerConfig) (*Runner, error)

NewRunner creates a Runner from the given config.

func (*Runner) AuthToken 

func (r *Runner) AuthToken() string

AuthToken returns the resolved bearer token. Empty if auth is disabled.

func (*Runner) AuthorizeURL 

func (r *Runner) AuthorizeURL(ctx context.Context, subject, server string) (string, error)

AuthorizeURL returns the delegated-consent login link for {subject, server} via the configured provider — standalone builds it locally, managed fetches it from the platform, an embedder may inject its own. The consent deliverer (e.g. Slack, #343) calls this to populate the prompt. Errors when no provider is wired (delivery then falls back to the mcp_auth_required audit event).

func (*Runner) CaptureStatedIntent 

func (r *Runner) CaptureStatedIntent(ctx context.Context, taskID string, msg *a2a.Message)

CaptureStatedIntent extracts the first user-authored text part from an inbound A2A message and registers it with the intent engine. Called from the tasks/send handlers immediately after CheckInbound admits the message. No-op when the engine is disabled or the message has no text.

Design note: takes the WHOLE message rather than a pre-extracted string so the choice of "which text counts as intent" stays here. Today we concatenate all text parts of the first message; a future refinement (structured "intent:" prefix, or an explicit A2A header) can change this without touching call sites.

func (*Runner) PublishConsentArtifact 

func (r *Runner) PublishConsentArtifact(taskID, subject, server, authorizeURL string, deadline time.Time)

PublishConsentArtifact writes the "Connect <server>" login link onto the parked task's auth-required artifact — a DURABLE record a UI/A2A client can render. It is the always-on backstop: channel delivery (Slack, #343) is an additive push on top of it, so a per-subject channel failure never leaves the user without a link. nil-safe (no task store / empty link ⇒ no-op).

func (*Runner) ResolveAuth 

func (r *Runner) ResolveAuth() error

ResolveAuth resolves the auth token early (before Run). This is needed so channel adapters can be configured with the token before Run() blocks. Safe to call multiple times — subsequent calls are no-ops.

Invariant: after this returns nil, EITHER r.authToken is non-empty OR r.cfg.NoAuth is true. resolveAuth() relies on this when it conditionally prepends the loopback static_token (review #10). If a future refactor adds a return path that violates this invariant, channel-adapter callbacks will silently break — the test TestResolveAuth_InvariantMintsTokenInNonNoAuthPath in auth_chain_test.go pins the property.

func (*Runner) Run 

func (r *Runner) Run(ctx context.Context) error

Run starts the development server. It blocks until ctx is cancelled.

func (*Runner) SetAuthorizeURLProvider 

func (r *Runner) SetAuthorizeURLProvider(fn AuthorizeURLProvider)

SetAuthorizeURLProvider overrides how the consent link is built. Standalone wires its own by default; a managed platform sets one that returns its own authorize URL (its client_id/state/redirect_uri) so Forge never constructs a managed URL from local config. Must be called before Run(). Takes precedence over the config-driven managed/standalone providers.

func (*Runner) SetCallbackCompleter 

func (r *Runner) SetCallbackCompleter(fn CallbackCompleter)

SetCallbackCompleter enables the STANDALONE loopback consent callback (#330): the injected func exchanges an OAuth code for a token and stores it for {subject, server}. When set, GET /mcp/oauth/callback is registered; when nil (managed mode), it is not — the platform hosts its own callback. Must be called before Run().

func (*Runner) SetConsentDeliverer 

func (r *Runner) SetConsentDeliverer(fn ConsentDeliverer)

SetConsentDeliverer sets the callback used to deliver MCP auth-required consent prompts (#330) — the managed platform injects one that hands off to its consent flow; standalone leaves it nil until the loopback resolver lands. Must be called before Run().

func (*Runner) SetDeferralNotifier 

func (r *Runner) SetDeferralNotifier(fn DeferralNotifier)

SetDeferralNotifier sets the callback used to deliver DEFER (R4c) approval requests to channel adapters (#310). Must be called before Run().

func (*Runner) SetScheduleNotifier 

func (r *Runner) SetScheduleNotifier(fn ScheduleNotifier)

SetScheduleNotifier sets the callback used to deliver scheduled task results to channel adapters. Must be called before Run().

func (*Runner) SetStatus 

func (r *Runner) SetStatus(id string, s a2a.TaskStatus) a2a.TaskStatus

registerDeferHook wires the R4c (#211) BeforeToolExec hook that pauses the executor when a tool is listed in `security.defer.tools`.

Pause mechanism: the hook goroutine (which is holding the HTTP request open in the tasks/send path) blocks on Handle.WaitCtx. While blocked:

  • The task's Status in the store flips to `deferred` so parallel `GET /tasks/{id}` requests see the deferred state.
  • A `task_deferred` audit event is emitted.
  • The timeout timer runs in the background.

When the deferral resolves (via POST /tasks/{id}/decisions or the timeout):

  • approve → hook returns nil; the tool proceeds; task status flips back to `working`.
  • reject → hook returns an error; the tool fails with a defer-denied message; task ends `failed`.
  • timeout → same as reject with a distinct audit event.

SetStatus lets *Runner satisfy TaskStatusStore so the defer hook can flip the task's Status via the runner even before the a2a server (which owns the store) has been constructed. Resolves r.taskStore lazily — nil-safe if the runner is being used outside a real server (unit tests).

type RunnerConfig 

type RunnerConfig struct {
	Config            *types.ForgeConfig
	WorkDir           string
	Port              int
	Host              string        // bind host (e.g. "127.0.0.1" for serve, "" for run)
	ShutdownTimeout   time.Duration // graceful shutdown timeout (0 = immediate)
	MockTools         bool
	EnforceGuardrails bool
	ModelOverride     string
	ProviderOverride  string
	EnvFilePath       string
	Verbose           bool
	Channels          []string // active channel adapters from --with flag
	NoAuth            bool     // disable bearer token authentication
	AuthToken         string   // explicit bearer token (empty = auto-generate)
	AuthURL           string   // external auth provider URL for token validation
	AuthOrgID         string   // org_id sent to external auth provider
	CORSOrigins       []string // CORS allowed origins (from --cors-origins flag)

	// AuditExport configures the FWS-7 audit export sinks (Unix socket
	// or localhost HTTP fallback). Zero value = pre-FWS-7 behavior
	// (stderr only). See issue #95.
	AuditExport coreruntime.AuditExportConfig

	// AuditPayloadCapture is the opt-in raw-payload capture for audit
	// events: LLM messages / completions, tool args / results. All
	// flags default off (metadata-only audit). See issue #91 / FWS-8
	// and docs/security/audit-logging.md#payload-capture-fws-8.
	AuditPayloadCapture coreruntime.AuditPayloadCapture

	// RateLimitOverride carries CLI-flag-derived overrides for the
	// per-IP A2A rate limiter. Nil = no CLI overrides; the resolver
	// will fall through to FORGE_RATE_LIMIT_* env vars and
	// cfg.Server.RateLimit before defaulting to the FWS-10 baseline.
	// See issue #110 / FWS-10.
	RateLimitOverride *RateLimitOverride

	// TracingFlags carries CLI-flag-derived OTel tracing overrides.
	// Zero value = "no CLI overrides"; the runner's tracing resolver
	// falls through to env (OTEL_*) and the
	// observability.tracing block of forge.yaml. See issue #103 / OTel
	// Tracing v1 (initiative #108).
	TracingFlags TracingFlags

	// RuntimeVersion is the Forge cli's own build version. Used for
	// the `forge.runtime.version` OTel resource attribute so backends
	// can compare agent runs across Forge upgrade waves. Empty = "dev".
	RuntimeVersion string

	// RuntimeCommit is the Forge cli's own build commit (short SHA),
	// injected via `-X main.commit`. Shown on the startup banner next to
	// RuntimeVersion so a running agent's exact binary is identifiable.
	// "none"/"" = unset (a dev build) → the banner shows just the version.
	RuntimeCommit string
}

RunnerConfig holds configuration for the Runner.

type ScheduleNotifier 

type ScheduleNotifier func(ctx context.Context, channel, target string, response *a2a.Message) error

ScheduleNotifier is called after a scheduled task completes to deliver the result to the appropriate channel (e.g. Slack, Telegram).

type StubExecutor 

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

StubExecutor implements AgentExecutor by returning an error indicating that no LLM configuration is available. Used as a fallback when no provider is configured for a custom framework agent.

func NewStubExecutor 

func NewStubExecutor(framework string) *StubExecutor

NewStubExecutor creates a StubExecutor for the given framework name.

func (*StubExecutor) Close 

func (s *StubExecutor) Close() error

Close is a no-op for StubExecutor.

func (*StubExecutor) Execute 

func (s *StubExecutor) Execute(ctx context.Context, task *a2a.Task, msg *a2a.Message) (*a2a.Message, error)

Execute returns an error indicating execution is not configured.

func (*StubExecutor) ExecuteStream 

func (s *StubExecutor) ExecuteStream(ctx context.Context, task *a2a.Task, msg *a2a.Message) (<-chan *a2a.Message, error)

ExecuteStream returns an error indicating execution is not configured.

type SubprocessExecutor 

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

SubprocessExecutor wraps a SubprocessRuntime to implement AgentExecutor. It delegates to the runtime's Invoke/Stream methods and extracts the message from the returned task.

func NewSubprocessExecutor 

func NewSubprocessExecutor(rt *SubprocessRuntime) *SubprocessExecutor

NewSubprocessExecutor creates an executor that delegates to the given runtime.

func (*SubprocessExecutor) Close 

func (s *SubprocessExecutor) Close() error

Close is a no-op; the subprocess lifecycle is managed by SubprocessRuntime.

func (*SubprocessExecutor) Execute 

func (s *SubprocessExecutor) Execute(ctx context.Context, task *a2a.Task, msg *a2a.Message) (*a2a.Message, error)

Execute calls the runtime's Invoke and extracts the status message.

func (*SubprocessExecutor) ExecuteStream 

func (s *SubprocessExecutor) ExecuteStream(ctx context.Context, task *a2a.Task, msg *a2a.Message) (<-chan *a2a.Message, error)

ExecuteStream calls the runtime's Stream and converts task updates to messages.

type SubprocessRuntime 

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

SubprocessRuntime manages a child agent process and proxies A2A requests to it.

func NewSubprocessRuntime 

func NewSubprocessRuntime(entrypoint, workDir string, env map[string]string, logger coreruntime.Logger) *SubprocessRuntime

NewSubprocessRuntime creates a runtime that will start the given entrypoint command, passing PORT as an env var for the subprocess to listen on.

func (*SubprocessRuntime) Healthy 

func (s *SubprocessRuntime) Healthy(ctx context.Context) bool

Healthy checks if the subprocess is responding.

func (*SubprocessRuntime) Invoke 

func (s *SubprocessRuntime) Invoke(ctx context.Context, taskID string, msg *a2a.Message) (*a2a.Task, error)

Invoke sends a synchronous tasks/send request to the subprocess.

func (*SubprocessRuntime) Restart 

func (s *SubprocessRuntime) Restart(ctx context.Context) error

Restart stops and re-starts the subprocess.

func (*SubprocessRuntime) Start 

func (s *SubprocessRuntime) Start(ctx context.Context) error

Start launches the subprocess and waits for it to become healthy.

func (*SubprocessRuntime) Stop 

func (s *SubprocessRuntime) Stop() error

Stop sends SIGTERM, waits 5s, then SIGKILL if needed.

func (*SubprocessRuntime) Stream 

func (s *SubprocessRuntime) Stream(ctx context.Context, taskID string, msg *a2a.Message) (<-chan *a2a.Task, error)

Stream sends a tasks/sendSubscribe request. If the subprocess returns SSE, events are streamed; otherwise the response is wrapped as a single item.

type TaskStatusStore 

type TaskStatusStore interface {
	// SetStatus replaces the status of task `id`. Returns the
	// previous status so callers can restore it after a resolve.
	SetStatus(id string, s a2a.TaskStatus) a2a.TaskStatus
}

TaskStatusStore is the narrow interface the defer hook uses to flip task status while blocked. The runner's a2a task store satisfies this; the interface exists so the hook doesn't take a hard dep on the full server.TaskStore surface (which would drag heavy imports into forge-cli/runtime).

type TracingFlags 

type TracingFlags struct {
	Enabled        *bool
	Endpoint       *string
	Protocol       *string
	Sampler        *string
	SamplerRatio   *float64
	Timeout        *time.Duration
	ServiceName    *string
	CaptureContent *bool
	Redact         *bool
}

TracingFlags carries the CLI-flag-derived tracing overrides. Pointers so the resolver can distinguish "flag not passed" from "explicitly zero" — `--otel-sampler-ratio 0` is a legitimate ask (drop everything), distinct from "no --otel-sampler-ratio flag".

Populated by forge-cli/cmd/run.go from the `--otel-*` flag variables; passed through RunnerConfig.TracingFlags into ResolveTracingConfig. Nil = "no CLI overrides" — equivalent to a zero-value struct.

type Verdict 

type Verdict struct {
	Decision coreruntime.PolicyDecision
	Reason   string
	Op       string
	Defer    *deferengine.Spec // set only when Decision == DecisionDefer
}

Verdict is the resolver output at the CLI enforcement layer. It carries a full deferengine.Spec (which includes Approvers) rather than runtime.DeferSpec (which does not), so a PDP defer verdict feeds the parking machinery directly.

CONTRACT: a zero-value Verdict is UNDEFINED and must never be treated as a decision. Note DecisionAllow is the zero value of PolicyDecision, so an accidentally-unconstructed Verdict{} would read as allow — every Verdict must be constructed with an explicit Decision, and the enforcement hook's default arm fails closed as a backstop. Do not add a code path that returns a Verdict{} on an authorization boundary.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.