Method Run

internal/inferexec/orchestrator.go:45–216

Run executes normalize → policy → (agent stub?) → admission → route → (RAG?) → backend → finalize.

(ctx context.Context, in RunInput)

// Run executes normalize → policy → (agent stub?) → admission → route → (RAG?) → backend → finalize.
func (o *Orchestrator) Run(ctx context.Context, in RunInput) *Result {
	_ = ExecutionPlanForRequestType(in.Req.RequestType)

	req := in.Req
	requestID := in.RequestID
	sw := in.SW
	emit := in.EmitRequestTrace
	if emit == nil {
		emit = func(string, string) {}
	}

	ledgerAt := time.Now()
	if o.Ledger != nil {
		o.Ledger.CreateRequest(ctx, in.TraceID, requestID, *req, ledgerAt)
	}

	_ = sw.Run(ctx, execution.StepNormalize, "", map[string]any{
		"request_type": req.RequestType, "pipeline_ref": req.PipelineRef,
	}, func() (map[string]any, error) {
		return map[string]any{"ok": true}, nil
	})

	policyDecision, err := o.Policy.Evaluate(ctx, *req)
	if err != nil {
		o.failLedger(ctx, requestID)
		if st, ec, ok := InferBudgetHTTPStatus(ctx, err); ok {
			if o.NoteTimeout != nil {
				o.NoteTimeout(err)
			}
			emit("", "gateway_timeout")
			return &Result{Failure: &Failure{HTTPStatus: st, ErrorCode: ec, Message: "inference request deadline exceeded", TraceResult: "gateway_timeout"}}
		}
		emit("", "policy_error")
		return &Result{Failure: &Failure{HTTPStatus: http.StatusInternalServerError, ErrorCode: "policy_error", Message: err.Error(), TraceResult: "policy_error"}}
	}

	if err := sw.Run(ctx, execution.StepPolicyCheck, "", map[string]any{"tenant_id": req.TenantID}, func() (map[string]any, error) {
		if !policyDecision.Allowed {
			return map[string]any{"allowed": false, "reason": policyDecision.Reason}, fmt.Errorf("policy rejected: %s", policyDecision.Reason)
		}
		return map[string]any{"allowed": true, "reason": policyDecision.Reason}, nil
	}); err != nil {
		log.Printf("event=policy_rejected request_id=%s tenant_id=%s reason=%q", requestID, req.TenantID, policyDecision.Reason)
		o.failLedger(ctx, requestID)
		emit("", "policy_rejected")
		return &Result{Failure: &Failure{HTTPStatus: http.StatusTooManyRequests, ErrorCode: "policy_rejected", Message: policyDecision.Reason, TraceResult: "policy_rejected"}}
	}
	*req = policyDecision.Normalized

	if req.RequestType == types.RequestTypeAgent {
		_ = sw.Run(ctx, execution.StepAgentStub, "", map[string]any{"pipeline": req.PipelineRef}, func() (map[string]any, error) {
			return nil, fmt.Errorf("agent execution not implemented")
		})
		o.failLedger(ctx, requestID)
		emit("", "agent_not_implemented")
		return &Result{Failure: &Failure{HTTPStatus: http.StatusNotImplemented, ErrorCode: "agent_not_implemented", Message: "agent execution is not implemented in this release", TraceResult: "agent_not_implemented"}}
	}