Documentation
¶
Overview ¶
Package researchexecute implements the execute_subqueries component from ADR-045 Phase 1 (PR 4 of six per docs/operations/22-adr045-phase1-plan.md).
execute_subqueries is the code-heavy stage of the graph-search rule chain. It receives a publish trigger on component.execute_subqueries.<loop_id>, reads the upstream research.Intent + research.RouteDecision payloads from AGENT_LOOPS, materializes one or more typed sub-queries from the routing intent, executes them in parallel across multiple retrieval tiers, normalises + dedups results, enforces the caller's token budget, and writes a fusion.Evidence array envelope plus an execute.complete.<loop_id> trigger key that R3 watches to dispatch the assess_sufficiency stage.
Two input shapes — same component:
walk_seeds args: model emits seed references (name/partial_id/candidate_index); component resolves to full 6-part federated entity IDs via the entity index, then dispatches multi-hop expansion as predicate_walk sub-queries.
decompose args: model emits decomposition intent (axes/focus/scope); component materializes typed sub-queries from MINIMAL TEMPLATES — entity_type → entity_state, time → temporal_range, predicate → predicate_walk (the fallback for axes that don't map to a dedicated primitive). spatial is deferred to Phase 2 (graph-index-spatial wire).
Architectural notes:
Pure-code component (no LLM calls). All work is graph + index retrieval against existing gateways.
Tier 0 (predicate queries) executes via graph-query NATS- direct subjects (graph.query.entitiesByPrefix etc.). Tier 1 (BM25) executes via graph.query.searchGraph (same surface research-graph-classify uses for initial candidate retrieval). Tier 2 (neural) is deferred to Phase 2.
Sub-query types live in pkg/fusion (generic leaf package). Reshapeable without cross-package churn if Phase 2 learns better primitives. Type aliases in subquery.go keep this package's call sites unchanged.
Parallel fan-out via errgroup with a bounded concurrency cap (config-driven). Per-tier ordering with recency tie-break; learned ranker deferred to Phase 2.
Provenance preserved end-to-end — every Evidence carries {tier, source, entity_id} the agent can quote back. No fabricated refs.
All public methods safe for concurrent use across loops; the component holds no per-call mutable state. Same pattern as research-graph-classify and research-graph-route.
Index ¶
- Constants
- func NewProcessor(rawConfig json.RawMessage, deps component.Dependencies) (component.Discoverable, error)
- func Register(registry *component.Registry) error
- type BM25Args
- type Component
- func (c *Component) ConfigSchema() component.ConfigSchema
- func (c *Component) DataFlow() component.FlowMetrics
- func (c *Component) Health() component.HealthStatus
- func (c *Component) Initialize() error
- func (c *Component) InputPorts() []component.Port
- func (c *Component) Meta() component.Metadata
- func (c *Component) OutputPorts() []component.Port
- func (c *Component) Start(ctx context.Context) error
- func (c *Component) Stop(timeout time.Duration) error
- type Config
- type EntityStateArgs
- type GraphQueryClient
- type LoopStore
- type PredicateWalkArgs
- type SubQuery
- type SubQueryType
- type TemporalRangeArgs
Constants ¶
const ( // DefaultExecuteTimeout caps the wall-clock for the full // fan-out (all sub-queries × all tiers). Generous default; per- // tier deadlines are derived as fractions of this cap. DefaultExecuteTimeout = 60 * time.Second // DefaultMaxParallelism caps concurrent sub-query execution. // Keeps fan-out from saturating the responding gateways under // a wide-decompose scenario; operators can raise for staging // environments with more capacity. DefaultMaxParallelism = 8 // DefaultMaxResultsPerSubquery caps per-sub-query result count // before ranking + budget enforcement run. Prevents a single // runaway sub-query from dominating the evidence array. DefaultMaxResultsPerSubquery = 50 // DefaultBudgetTokens is the fallback evidence-budget when the // upstream Intent doesn't supply one. Should never normally // fire (Intent.ResolvedBudgetTokens supplies a default of 4000 // per agentic/research) but acts as a safety net. DefaultBudgetTokens = 4000 // DefaultPerSubqueryTimeoutFraction is the fraction of the // overall ExecuteTimeout each sub-query gets as its per-call // deadline. 0.5 leaves slack for fan-out overhead and ranking; // operators can tighten. DefaultPerSubqueryTimeoutFraction = 0.5 )
Default knobs surfaced as exported constants so the materializer + fan-out tests and operator docs can reference them by name rather than duplicating literals.
const ( SubQueryTypeEntityState = fusion.SubQueryTypeEntityState SubQueryTypePredicateWalk = fusion.SubQueryTypePredicateWalk SubQueryTypeTemporalRange = fusion.SubQueryTypeTemporalRange SubQueryTypeBM25 = fusion.SubQueryTypeBM25 )
Re-export the SubQueryType constants from fusion so existing usages in this package and its tests compile unchanged.
const ComponentName = "research-graph-execute"
ComponentName is the canonical registry name + log subsystem.
Variables ¶
This section is empty.
Functions ¶
func NewProcessor ¶
func NewProcessor(rawConfig json.RawMessage, deps component.Dependencies) (component.Discoverable, error)
NewProcessor is the component-factory shape registered with the component registry.
Types ¶
type Component ¶
type Component struct {
// contains filtered or unexported fields
}
Component implements the execute_subqueries processor. Struct field set is intentionally small; lifecycle methods own the NATS plumbing and the per-message handler hands off to the pure executeAll function in handler.go.
func (*Component) ConfigSchema ¶
func (c *Component) ConfigSchema() component.ConfigSchema
ConfigSchema implements Discoverable.
func (*Component) DataFlow ¶
func (c *Component) DataFlow() component.FlowMetrics
DataFlow implements Discoverable.
func (*Component) Health ¶
func (c *Component) Health() component.HealthStatus
Health implements Discoverable.
func (*Component) Initialize ¶
Initialize is part of the LifecycleComponent contract — no pre- Start work.
func (*Component) InputPorts ¶
InputPorts implements Discoverable.
func (*Component) OutputPorts ¶
OutputPorts implements Discoverable. execute_subqueries has no NATS-publishing outputs; emits via KV writes to AGENT_LOOPS.
type Config ¶
type Config struct {
Ports *component.PortConfig `` /* 179-byte string literal not displayed */
LoopsBucket string `` /* 175-byte string literal not displayed */
ExecuteTimeout time.Duration `` /* 163-byte string literal not displayed */
MaxParallelism int `` /* 193-byte string literal not displayed */
MaxResultsPerSubquery int `` /* 252-byte string literal not displayed */
}
Config holds operator-tunable knobs for the execute_subqueries component.
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns a default Config skeleton with the standard execute_subqueries input port.
func (*Config) ApplyDefaults ¶
func (c *Config) ApplyDefaults()
ApplyDefaults fills in defaults for unset fields.
type EntityStateArgs ¶
type EntityStateArgs = fusion.EntityStateArgs
EntityStateArgs aliases fusion.EntityStateArgs.
type GraphQueryClient ¶
type GraphQueryClient = fusion.GraphQueryClient
GraphQueryClient is the narrow surface this component consumes from graph-query / graph-index. Production wraps NATS-direct subjects (graph.query.entitiesByPrefix, graph.query.searchGraph, etc.); tests substitute an in-memory fake so the matrix doesn't need a live graph stack.
This is a type alias for fusion.GraphQueryClient — the interface is defined in pkg/fusion; this alias keeps the component's imports tidy without an extra indirection for callers already working inside this package.
type LoopStore ¶
type LoopStore interface {
// GetIntent loads the research_intent payload from the
// research.request.received.<loopID> key.
GetIntent(ctx context.Context, loopID string) (*research.Intent, error)
// GetClassifierOutput loads the upstream ClassifierOutput from
// the classify.complete.<loopID> trigger key. Needed for
// walk_seeds candidate_index resolution + decompose entity_type
// anchoring.
GetClassifierOutput(ctx context.Context, loopID string) (*research.ClassifierOutput, error)
// GetRouteDecision loads the upstream RouteDecision from the
// route.complete.<loopID> trigger key. Drives sub-query
// materialization.
GetRouteDecision(ctx context.Context, loopID string) (*research.RouteDecision, error)
// PutExecutionOutput writes the ExecutionOutput envelope at
// R3's trigger key execute.complete.<loopID>.
PutExecutionOutput(ctx context.Context, loopID string, envelope []byte) error
// PutSnapshot writes the envelope at the stable non-trigger
// key execute.snapshot.<loopID> so operators / downstream
// queryability can read without racing R3's wildcard watcher.
PutSnapshot(ctx context.Context, loopID string, envelope []byte) error
}
LoopStore is the AGENT_LOOPS read/write surface this component consumes. Production wraps natsclient.KVStore; tests substitute an in-memory map.
type PredicateWalkArgs ¶
type PredicateWalkArgs = fusion.PredicateWalkArgs
PredicateWalkArgs aliases fusion.PredicateWalkArgs.
type SubQueryType ¶
type SubQueryType = fusion.SubQueryType
SubQueryType aliases fusion.SubQueryType.
type TemporalRangeArgs ¶
type TemporalRangeArgs = fusion.TemporalRangeArgs
TemporalRangeArgs aliases fusion.TemporalRangeArgs.
Source Files