execution

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Jul 14, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ToolHookStagePre  = runtimehooks.ToolHookStagePre
	ToolHookStagePost = runtimehooks.ToolHookStagePost
)
View Source 
const DefaultEventQueueCapacity = 1000

DefaultEventQueueCapacity is the default buffer size for EventQueue.

View Source 
const DefaultRuntimeEventQueueCapacity = 1000

DefaultRuntimeEventQueueCapacity is the default buffer size for RuntimeEventQueue.

Variables

This section is empty.

Functions

This section is empty.

Types

type ErrorStage 

type ErrorStage string

ErrorStage represents when an error occurred. 

const (
	ErrorStagePermission ErrorStage = "permission"
	ErrorStageExecution  ErrorStage = "execution"
	ErrorStageTimeout    ErrorStage = "timeout"
	ErrorStageDisabled   ErrorStage = "disabled"
	ErrorStageHook       ErrorStage = "hook"
)

type EventQueue 

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

EventQueue is a non-blocking buffered channel for streaming APIResponseChunks.

Emitters call Emit; readers consume Recv(). When the queue is full, Emit drops the chunk and increments OverflowCount so callers can detect backpressure. Close must be called once to signal readers that no more chunks are coming.

func NewEventQueue 

func NewEventQueue(capacity int) *EventQueue

NewEventQueue creates an EventQueue with the given buffer capacity. If capacity <= 0, DefaultEventQueueCapacity is used.

func (*EventQueue) Close 

func (q *EventQueue) Close()

Close signals readers that no more chunks will arrive by closing the channel. It is safe to call Close multiple times; only the first call has effect.

func (*EventQueue) Drain 

func (q *EventQueue) Drain(ctx context.Context) error

Drain blocks until the queue's internal buffer is empty or ctx is cancelled. It does NOT consume chunks — the caller's reader goroutine must still drain Recv(). Drain is useful in tests and shutdown sequences where the caller wants to wait until the consumer has caught up before asserting on results.

Returns ctx.Err() if the context expires before the buffer empties, nil otherwise.

func (*EventQueue) Emit 

func (q *EventQueue) Emit(chunk types.APIResponseChunk)

Emit sends chunk into the queue. If the queue is full or already closed, the chunk is discarded and the overflow counter is incremented.

func (*EventQueue) EmitBlocking 

func (q *EventQueue) EmitBlocking(ctx context.Context, chunk types.APIResponseChunk) (ok bool)

EmitBlocking sends chunk into the queue, blocking until the queue has room or until ctx is cancelled. Returns true if the chunk was enqueued, false if ctx was cancelled before the chunk could be sent (chunk is dropped in that case).

Use EmitBlocking instead of Emit when losing chunks is not acceptable, for example when a caller requires complete transcript delivery. Be aware that blocking here stalls the engine loop — the consumer must drain promptly.

func (*EventQueue) EmittedCount 

func (q *EventQueue) EmittedCount() int64

EmittedCount returns the total number of chunks successfully placed in the queue.

func (*EventQueue) OverflowCount 

func (q *EventQueue) OverflowCount() int64

OverflowCount returns the number of chunks dropped because the queue was full.

func (*EventQueue) Recv 

func (q *EventQueue) Recv() <-chan types.APIResponseChunk

Recv returns the read-only channel. Readers range over it or select from it. The channel is closed when Close is called.

func (*EventQueue) Stats 

func (q *EventQueue) Stats() EventQueueStats

Stats returns a point-in-time snapshot of queue counters. Pending is the number of chunks buffered but not yet read by the consumer.

type EventQueueStats 

type EventQueueStats struct {
	Emitted  int64
	Overflow int64
	Pending  int // chunks currently buffered (not yet consumed)
	Closed   bool
}

EventQueueStats holds a point-in-time snapshot of EventQueue counters.

type ExecuteRequest 

type ExecuteRequest struct {
	ToolUses []types.ToolUseContent `json:"tool_uses"`
	Tools    map[string]tool.Tool   `json:"-"`

	PermissionCheck    types.CanUseToolFn         `json:"-"`
	PermissionResolver types.PermissionResolver   `json:"-"`
	PermissionContext  *types.PermissionContext   `json:"-"`
	SessionID          types.SessionID            `json:"session_id"`
	TurnID             types.TurnID               `json:"turn_id"`
	WorkingDirectory   string                     `json:"working_directory,omitempty"`
	PermissionMode     types.PermissionMode       `json:"permission_mode"`
	ProgressCallback   func(types.ToolProgress)   `json:"-"`
	DenialTracking     *types.DenialTrackingState `json:"-"`
	Transcript         []types.Message            `json:"-"`

	// EnableSandbox indicates whether tools should run in sandbox mode.
	// Used for sandbox auto-allow permission logic.
	EnableSandbox bool `json:"enable_sandbox,omitempty"`

	// ShouldAvoidPermissionPrompts is true for headless/background agents
	// that cannot display UI prompts.
	ShouldAvoidPermissionPrompts bool `json:"should_avoid_permission_prompts,omitempty"`
}

ExecuteRequest represents a request to execute tools.

type ExecuteResult 

type ExecuteResult struct {
	Results          []tool.CallResult    `json:"results"`
	Messages         []types.Message      `json:"messages"`
	Traces           []ToolExecutionTrace `json:"traces"`
	Errors           []ExecutionError     `json:"errors"`
	TotalDuration    time.Duration        `json:"total_duration_ms"`
	ProgressUpdates  []types.ToolProgress `json:"progress_updates"`
	FinalToolContext tool.ToolUseContext  `json:"-"`
}

ExecuteResult represents the result of executing tools.

type ExecutionError 

type ExecutionError struct {
	ToolUseID string     `json:"tool_use_id"`
	ToolName  string     `json:"tool_name"`
	Error     error      `json:"error"`
	Stage     ErrorStage `json:"stage"`
}

ExecutionError represents an error during tool execution.

type Orchestrator 

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

Orchestrator manages parallel tool execution. Pipeline per tool use (aligned with OpenClaude's runToolUse):

  1. resolve tool + IsEnabled check
  2. ValidateInput
  3. BackfillInput (observable enrichment for hooks/permissions)
  4. Pre-tool hooks
  5. Safety checks (bypass-immune)
  6. Permission pipeline (deny rules → local check → always-allow → global check)
  7. Denial tracking (auto mode)
  8. tool.Call()
  9. Post-tool hooks
  10. FormatResult (tool-controlled serialisation)
  11. Content replacement (max result size)
  12. Context modifier (serial only, or batch-ordered after concurrent)

func NewOrchestrator 

func NewOrchestrator() *Orchestrator

NewOrchestrator creates a new tool execution orchestrator.

func (*Orchestrator) AddHook 

func (o *Orchestrator) AddHook(hook ToolHook)

AddHook registers a pre- or post-tool-use hook.

func (*Orchestrator) Execute 

func (o *Orchestrator) Execute(ctx context.Context, req ExecuteRequest) (ExecuteResult, error)

Execute executes tools with proper parallelization.

func (*Orchestrator) SetDenialLimitConfig 

func (o *Orchestrator) SetDenialLimitConfig(config types.DenialLimitConfig)

SetDenialLimitConfig configures denial-based fallback behavior.

func (*Orchestrator) SetMaxConcurrency 

func (o *Orchestrator) SetMaxConcurrency(max int)

SetMaxConcurrency sets the maximum concurrency.

func (*Orchestrator) SetMonitoring 

func (o *Orchestrator) SetMonitoring(monitoringSys *monitoring.System)

SetMonitoring sets the monitoring system

func (*Orchestrator) SetSafetyChecker 

func (o *Orchestrator) SetSafetyChecker(checker types.SafetyChecker)

SetSafetyChecker sets the bypass-immune safety checker.

type RuntimeEventQueue 

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

RuntimeEventQueue is a non-blocking buffered channel for structured runtime events.

func NewRuntimeEventQueue 

func NewRuntimeEventQueue(capacity int) *RuntimeEventQueue

NewRuntimeEventQueue creates a RuntimeEventQueue with the given buffer capacity. If capacity <= 0, DefaultRuntimeEventQueueCapacity is used.

func (*RuntimeEventQueue) Close 

func (q *RuntimeEventQueue) Close()

Close signals readers that no more events will arrive.

func (*RuntimeEventQueue) Drain 

func (q *RuntimeEventQueue) Drain(ctx context.Context) error

Drain blocks until the queue's internal buffer is empty or ctx is cancelled.

func (*RuntimeEventQueue) Emit 

func (q *RuntimeEventQueue) Emit(event types.RuntimeEvent)

Emit sends an event into the queue. If the queue is full or already closed, the event is discarded and the overflow counter is incremented.

func (*RuntimeEventQueue) EmitBlocking 

func (q *RuntimeEventQueue) EmitBlocking(ctx context.Context, event types.RuntimeEvent) bool

EmitBlocking sends an event into the queue, blocking until the queue has room or until ctx is cancelled.

func (*RuntimeEventQueue) EmittedCount 

func (q *RuntimeEventQueue) EmittedCount() int64

EmittedCount returns the total number of events successfully placed in the queue.

func (*RuntimeEventQueue) OverflowCount 

func (q *RuntimeEventQueue) OverflowCount() int64

OverflowCount returns the number of events dropped because the queue was full.

func (*RuntimeEventQueue) Recv 

func (q *RuntimeEventQueue) Recv() <-chan types.RuntimeEvent

Recv returns the read-only runtime event channel.

func (*RuntimeEventQueue) Stats 

func (q *RuntimeEventQueue) Stats() RuntimeEventQueueStats

Stats returns a point-in-time snapshot of queue counters.

type RuntimeEventQueueStats 

type RuntimeEventQueueStats struct {
	Emitted  int64
	Overflow int64
	Pending  int
	Closed   bool
}

RuntimeEventQueueStats holds a point-in-time snapshot of RuntimeEventQueue counters.

type StreamingExecutionResult 

type StreamingExecutionResult struct {
	ToolUse    types.ToolUseContent
	Index      int
	Result     tool.CallResult
	Messages   []types.Message
	Error      error
	ErrorStage ErrorStage
	Progress   []types.ToolProgress
	Trace      ToolExecutionTrace
}

StreamingExecutionResult is the public snapshot returned by StreamingExecutor. Messages contains only the extra runtime messages emitted around the tool call; the canonical tool_result message is rebuilt by the engine when merging paths.

type StreamingExecutor 

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

func NewStreamingExecutor 

func NewStreamingExecutor(
	ctx context.Context,
	orchestrator *Orchestrator,
	req ExecuteRequest,
	tools map[string]tool.Tool,
	canUseTool types.CanUseToolFn,
	toolUseContext tool.ToolUseContext,
) *StreamingExecutor

func (*StreamingExecutor) Discard 

func (s *StreamingExecutor) Discard()

Discard cancels the derived context (so in-flight Execute calls return promptly), marks the executor aborted, waits for all goroutines to exit, and records a Canceled outcome for every tool that never got to run. After Discard returns, GetAllCompletedResults contains one entry per tool that was submitted — either a real result or a Canceled error.

func (*StreamingExecutor) DrainForAbort 

func (s *StreamingExecutor) DrainForAbort() []StreamingExecutionResult

DrainForAbort mirrors openclaude's getRemainingResults: it aborts all in-flight and pending tools and returns one StreamingExecutionResult per submitted tool — real result if the tool finished before the abort, or a Canceled error for anything still in-flight. Callers use this to build synthetic tool_result messages that keep the transcript consistent.

func (*StreamingExecutor) GetAllCompleted 

func (s *StreamingExecutor) GetAllCompleted() []toolExecutionOutcome

func (*StreamingExecutor) GetAllCompletedResults 

func (s *StreamingExecutor) GetAllCompletedResults() []StreamingExecutionResult

func (*StreamingExecutor) GetCompletedResults 

func (s *StreamingExecutor) GetCompletedResults() []toolExecutionOutcome

func (*StreamingExecutor) GetPendingCount 

func (s *StreamingExecutor) GetPendingCount() int

func (*StreamingExecutor) IsAborted 

func (s *StreamingExecutor) IsAborted() bool

func (*StreamingExecutor) SubmitToolUse 

func (s *StreamingExecutor) SubmitToolUse(toolUse types.ToolUseContent, index int)

func (*StreamingExecutor) SubmitToolUseChunk 

func (s *StreamingExecutor) SubmitToolUseChunk(chunk ToolUseChunk)

func (*StreamingExecutor) Wait 

func (s *StreamingExecutor) Wait(ctx context.Context) error

Wait blocks until all submitted tools complete or ctx is canceled. On cancellation Discard is called and ctx.Err() is returned, giving callers the same "drain for abort" semantics as openclaude's getRemainingResults.

type ToolExecutionTrace 

type ToolExecutionTrace struct {
	ToolUseID string `json:"tool_use_id"`
	ToolName  string `json:"tool_name"`

	ValidatedInput  map[string]any `json:"validated_input,omitempty"`
	BackfilledInput map[string]any `json:"backfilled_input,omitempty"`
	FinalInput      map[string]any `json:"final_input,omitempty"`

	LocalPermission  types.PermissionResult `json:"local_permission"`
	GlobalPermission types.PermissionResult `json:"global_permission"`

	Metadata map[string]any `json:"metadata,omitempty"`
}

ToolExecutionTrace captures the main execution stages for a single tool use.

type ToolHook 

type ToolHook = runtimehooks.ToolHook

type ToolHookInput 

type ToolHookInput = runtimehooks.ToolHookInput

type ToolHookResult 

type ToolHookResult = runtimehooks.ToolHookResult

type ToolHookStage 

type ToolHookStage = runtimehooks.ToolHookStage

type ToolHookStop 

type ToolHookStop = runtimehooks.ToolHookStop

type ToolUseChunk 

type ToolUseChunk struct {
	ToolUse   types.ToolUseContent
	Index     int
	IsPartial bool
}

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.