runtime

package
v0.1.10 Latest Latest
Warning

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

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

Documentation

Overview

Package runtime —— runtime 模块占位。

阶段 2 由对应 subagent 填充：handler.go / service.go / dto.go 等。 见 docs/13-phase1-prd.md 和 docs/14-week1-day-by-day.md。

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func StartEndpointRunWorker 

func StartEndpointRunWorker(ctx context.Context, svc *Service, cfg EndpointRunWorkerConfig)

StartEndpointRunWorker closes abandoned direct_http / mcp_server runs so an API process crash after run creation cannot leave user-visible calls stuck.

func StartRuntimePullRunWorker 

func StartRuntimePullRunWorker(ctx context.Context, svc *Service, cfg RuntimePullRunWorkerConfig)

StartRuntimePullRunWorker closes abandoned queued runtime runs so a crashed or misconfigured local worker cannot leave user-visible calls stuck forever.

Types

type AgentA2AContext 

type AgentA2AContext struct {
	CurrentRunID      string   `json:"current_run_id"`
	ParentRunID       string   `json:"parent_run_id,omitempty"`
	CallerAgentID     string   `json:"caller_agent_id,omitempty"`
	ProtocolContextID string   `json:"protocol_context_id,omitempty"`
	ProtocolTaskID    string   `json:"protocol_task_id,omitempty"`
	RootContextID     string   `json:"root_context_id,omitempty"`
	ParentContextID   string   `json:"parent_context_id,omitempty"`
	ParentTaskID      string   `json:"parent_task_id,omitempty"`
	TraceID           string   `json:"trace_id,omitempty"`
	ReferenceTaskIDs  []string `json:"reference_task_ids,omitempty"`
	CallAgentEndpoint string   `json:"call_agent_endpoint"`
	CallAgentMethod   string   `json:"call_agent_method"`
	AgentTokenType    string   `json:"agent_token_type"`
	AgentScopes       []string `json:"agent_scopes"`
}

AgentA2AContext tells an Agent how to delegate from its current run without making a human copy/paste a parent run id from the UI.

type AgentError 

type AgentError struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

AgentError 创作者侧错误（业务级 4xx 等场景）。

type AgentEvent 

type AgentEvent struct {
	EventType string                 `json:"event_type"`
	Payload   map[string]interface{} `json:"payload"`
}

AgentEvent 是 Agent endpoint 可选返回的运行中事件。

Phase 2 先允许少量 OpenLinker-native 事件；后续可映射到 A2A Message / Part / Artifact。

type AgentHeartbeatResponse 

type AgentHeartbeatResponse struct {
	AgentID                          string     `json:"agent_id"`
	AvailabilityStatus               string     `json:"availability_status"`
	LastCheckedAt                    *time.Time `json:"last_checked_at,omitempty"`
	ConsecutiveFailures              int32      `json:"consecutive_failures"`
	PendingRunCount                  int32      `json:"pending_run_count"`
	ClaimNow                         bool       `json:"claim_now"`
	NextClaimAfterSeconds            int32      `json:"next_claim_after_seconds"`
	RecommendedHeartbeatAfterSeconds int32      `json:"recommended_heartbeat_after_seconds"`
	MaxClaimWaitSeconds              int32      `json:"max_claim_wait_seconds"`
}

AgentHeartbeatResponse confirms that an Agent-bound access token owner is alive.

type AgentRequest 

type AgentRequest struct {
	Input         map[string]interface{} `json:"input"`
	Metadata      map[string]interface{} `json:"metadata,omitempty"`
	RunID         string                 `json:"run_id"`
	ParentRunID   string                 `json:"parent_run_id,omitempty"`
	CallerAgentID string                 `json:"caller_agent_id,omitempty"`
	A2A           *AgentA2AContext       `json:"a2a,omitempty"`
	Conversation  *ConversationContext   `json:"conversation,omitempty"`
}

AgentRequest 平台 → 创作者 endpoint 的请求体。

RunID 让创作者侧可对账 / 排查；Metadata 原样转发。

type AgentResponse 

type AgentResponse struct {
	Output   map[string]interface{} `json:"output"`
	Events   []AgentEvent           `json:"events,omitempty"`
	CostUSD  *float64               `json:"cost_usd,omitempty"`
	Metadata map[string]interface{} `json:"metadata,omitempty"`
	Error    *AgentError            `json:"error,omitempty"`
}

AgentResponse 创作者 endpoint → 平台的响应体。

Output 业务结果（成功时必填）；Error 业务错误（失败时必填）； CostUSD 创作者透明告知本次调用真实成本（可选，目前平台不做核对，仅记录用）。

type ConversationContext 

type ConversationContext struct {
	ID                   string                `json:"id"`
	SessionKey           string                `json:"session_key"`
	ProtocolContextID    string                `json:"protocol_context_id,omitempty"`
	RootContextID        string                `json:"root_context_id,omitempty"`
	CurrentRunID         string                `json:"current_run_id"`
	CurrentProtocolTask  string                `json:"current_protocol_task_id,omitempty"`
	HistoryBeforeCurrent []ConversationMessage `json:"history_before_current,omitempty"`
	Truncated            bool                  `json:"truncated"`
	Source               string                `json:"source"`
}

ConversationContext is the Core-owned multi-run message context for a protocol/root context. Callers send the current message; Core packages prior persisted messages so runtimes do not trust client-supplied history.

type ConversationMessage 

type ConversationMessage struct {
	RunID         string                 `json:"run_id"`
	EventSequence *int32                 `json:"event_sequence,omitempty"`
	Role          string                 `json:"role"`
	Content       string                 `json:"content"`
	Payload       map[string]interface{} `json:"payload,omitempty"`
	CreatedAt     string                 `json:"created_at,omitempty"`
}

type Delegation 

type Delegation struct {
	ParentRunID   uuid.UUID
	CallerAgentID uuid.UUID
	Reason        string
}

Delegation describes an Agent-mediated child run executed within an active parent run.

type DeliveryEnqueuer 

type DeliveryEnqueuer interface {
	EnqueueIfDefault(ctx context.Context, run *db.Run) error
}

DeliveryEnqueuer 触发 run 完成后向用户的默认投递目标发送。

同 WebhookEnqueuer 用接口注入避免 runtime → delivery 硬依赖。 实现见 internal/delivery.Service.EnqueueIfDefault。

type EndpointLimiter 

type EndpointLimiter interface {
	// contains filtered or unexported methods
}

func NewRedisEndpointLimiter 

func NewRedisEndpointLimiter(client redis.UniversalClient, prefix string, timeout time.Duration) EndpointLimiter

type EndpointRunTimeoutConfig 

type EndpointRunTimeoutConfig struct {
	StaleAfter time.Duration
	BatchSize  int32
}

EndpointRunTimeoutConfig controls how long direct_http / mcp_server runs may stay running before the platform converts them to terminal timeout.

type EndpointRunWorkerConfig 

type EndpointRunWorkerConfig struct {
	Interval   time.Duration
	StaleAfter time.Duration
	RunTimeout time.Duration
	BatchSize  int32
}

type Handler 

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

Handler 调用执行 HTTP 入口。

func NewHandler 

func NewHandler(svc runtimeService, cfg ...*config.Config) *Handler

NewHandler 构造 Handler。cfg 可选（测试可省略）。

func (*Handler) ClaimRuntimePullRun 

func (h *Handler) ClaimRuntimePullRun(c echo.Context) (err error)

func (*Handler) GetRun 

func (h *Handler) GetRun(c echo.Context) error

GetRun 查询单条调用详情（仅 owner）。

func (*Handler) GetRunArtifacts 

func (h *Handler) GetRunArtifacts(c echo.Context) error

GetRunArtifacts 查询 run 持久化产物。只返回给 run owner。

func (*Handler) GetRunEvents 

func (h *Handler) GetRunEvents(c echo.Context) error

GetRunEvents 查询 run 事件流。SSE 接口后续会复用同一 service 方法。

func (*Handler) GetRunMessages 

func (h *Handler) GetRunMessages(c echo.Context) error

GetRunMessages 查询 run 的稳定消息回放。只返回给 run owner。

func (*Handler) PostAgentHeartbeat 

func (h *Handler) PostAgentHeartbeat(c echo.Context) error

func (*Handler) PostRun 

func (h *Handler) PostRun(c echo.Context) error

PostRun 调用 Agent。

同步等待创作者 endpoint 返回（Phase 1 不做流式 / 异步队列）。 失败 / 超时 / 取消 → status='failed' or 'timeout' or 'canceled'，已退款。

func (*Handler) PostRunAsync 

func (h *Handler) PostRunAsync(c echo.Context) error

PostRunAsync 启动异步调用，立即返回 run_id，调用结果通过 GET /runs/:id 或 SSE 查询。

func (*Handler) PostRunEvent 

func (h *Handler) PostRunEvent(c echo.Context) error

PostRunEvent 允许 Agent endpoint 在执行中上报 run event。

鉴权使用 Agent 注册时的 endpoint_auth_header；平台调用 Agent 时也会把同一 secret 放进 X-OpenLinker-Token，Agent 可用它回调本接口。

func (*Handler) PostRuntimePullResult 

func (h *Handler) PostRuntimePullResult(c echo.Context) error

func (*Handler) RegisterAgentRuntime 

func (h *Handler) RegisterAgentRuntime(api *echo.Group)

RegisterAgentRuntime mounts Agent-bound access-token endpoints used by runtime_pull Agents. 

POST /agent-runtime/heartbeat        Agent 主动上报存活
GET  /agent-runtime/runs/claim       Agent 拉取一个 pending run
POST /agent-runtime/runs/:id/result  Agent 回传终态结果
GET  /agent-runtime/ws               Agent 出站 WebSocket，平台实时下发 run

func (*Handler) RegisterProtected 

func (h *Handler) RegisterProtected(api *echo.Group, runMw, queryMw echo.MiddlewareFunc)

RegisterProtected 注册需要鉴权的端点，分别接收 /run 与 /runs/:id 的 middleware。 

POST /run            同步调用 Agent   —— runMw（JWT + 访问令牌混合）
POST /runs           异步启动调用     —— runMw（JWT + 访问令牌混合）
GET  /runs/:id       单条调用详情     —— queryMw（可按部署选择 JWT-only 或 hybrid）
GET  /runs/:id/events 调用事件流      —— queryMw（轮询）
GET  /runs/:id/artifacts 运行产物      —— queryMw
GET  /runs/:id/messages 运行消息回放    —— queryMw
GET  /runs/:id/stream 调用事件 SSE    —— queryMw
POST /runs/:id/events Agent 上报事件  —— X-OpenLinker-Token（不使用用户 JWT）

GET /runs 列表由 dashboard 模块（subagent-6a）提供，本模块不挂。

调用方若两条路由想共用同一个 middleware，传入相同实例即可。

func (*Handler) RuntimeWebSocket 

func (h *Handler) RuntimeWebSocket(c echo.Context) error

RuntimeWebSocket keeps an outbound Agent connection open so the platform can assign queued runs without waiting for the next long-poll claim.

func (*Handler) SetEndpointLimiter 

func (h *Handler) SetEndpointLimiter(limiter EndpointLimiter)

func (*Handler) StreamRunEvents 

func (h *Handler) StreamRunEvents(c echo.Context) error

StreamRunEvents 以 SSE 输出 run events。

已结束的 run 会回放事件后关闭；运行中的 run 会轮询等待新事件直到终态或客户端断开。

type ReportRunEventRequest 

type ReportRunEventRequest struct {
	EventType string                 `json:"event_type"`
	Payload   map[string]interface{} `json:"payload,omitempty"`
}

ReportRunEventRequest Agent endpoint -> OpenLinker 的运行中事件上报请求体。

type RunA2AContextRequest 

type RunA2AContextRequest struct {
	ProtocolContextID string   `json:"protocol_context_id,omitempty"`
	ProtocolTaskID    string   `json:"protocol_task_id,omitempty"`
	RootContextID     string   `json:"root_context_id,omitempty"`
	ParentContextID   string   `json:"parent_context_id,omitempty"`
	ParentTaskID      string   `json:"parent_task_id,omitempty"`
	ParentRunID       string   `json:"parent_run_id,omitempty"`
	CallerAgentID     string   `json:"caller_agent_id,omitempty"`
	TargetAgentID     string   `json:"target_agent_id,omitempty"`
	TraceID           string   `json:"trace_id,omitempty"`
	ReferenceTaskIDs  []string `json:"reference_task_ids,omitempty"`
	Source            string   `json:"source,omitempty"`
}

type RunA2AContextResponse 

type RunA2AContextResponse struct {
	ProtocolContextID string   `json:"protocol_context_id"`
	ProtocolTaskID    string   `json:"protocol_task_id"`
	RootContextID     string   `json:"root_context_id"`
	ParentContextID   string   `json:"parent_context_id,omitempty"`
	ParentTaskID      string   `json:"parent_task_id,omitempty"`
	ParentRunID       string   `json:"parent_run_id,omitempty"`
	CallerAgentID     string   `json:"caller_agent_id,omitempty"`
	TargetAgentID     string   `json:"target_agent_id,omitempty"`
	TraceID           string   `json:"trace_id,omitempty"`
	ReferenceTaskIDs  []string `json:"reference_task_ids,omitempty"`
	Source            string   `json:"source,omitempty"`
}

type RunArtifactResponse 

type RunArtifactResponse struct {
	ID               string                 `json:"id"`
	RunID            string                 `json:"run_id"`
	ArtifactType     string                 `json:"artifact_type"`
	Title            string                 `json:"title"`
	Content          map[string]interface{} `json:"content"`
	Visibility       string                 `json:"visibility"`
	SourceArtifactID string                 `json:"source_artifact_id,omitempty"`
	MimeType         string                 `json:"mime_type,omitempty"`
	FileURI          string                 `json:"file_uri,omitempty"`
	FileName         string                 `json:"file_name,omitempty"`
	FileSHA256       string                 `json:"file_sha256,omitempty"`
	FileSizeBytes    *int64                 `json:"file_size_bytes,omitempty"`
	CreatedAt        time.Time              `json:"created_at"`
}

RunArtifactResponse is a persisted, owner-readable artifact produced by a run.

type RunEventResponse 

type RunEventResponse struct {
	EventID     string                 `json:"event_id"`
	RunID       string                 `json:"run_id"`
	ParentRunID string                 `json:"parent_run_id,omitempty"`
	Sequence    int32                  `json:"sequence"`
	EventType   string                 `json:"event_type"`
	Payload     map[string]interface{} `json:"payload"`
	CreatedAt   time.Time              `json:"created_at"`
}

RunEventResponse GET /api/v1/runs/:id/events 响应项。

sequence 在单个 run 内单调递增，后续 SSE 可用作 Last-Event-ID。

type RunEvidenceSummary 

type RunEvidenceSummary struct {
	Status            string  `json:"status"`
	CoverageStatus    string  `json:"coverage_status"`
	MatchedSkillCount int     `json:"matched_skill_count"`
	MissingSkillCount int     `json:"missing_skill_count"`
	UsedMCPToolCount  int     `json:"used_mcp_tool_count"`
	ArtifactCount     int     `json:"artifact_count"`
	MessageCount      int     `json:"message_count"`
	DeliveryStatus    *string `json:"delivery_status,omitempty"`
	PublicSafe        bool    `json:"public_safe"`
	EvidenceURL       string  `json:"evidence_url"`
}

RunEvidenceSummary gives UI and external clients a compact view of why a run is trustworthy without forcing them to stitch together multiple endpoints.

type RunMessageResponse 

type RunMessageResponse struct {
	ID            string                 `json:"id"`
	RunID         string                 `json:"run_id"`
	EventSequence *int32                 `json:"event_sequence,omitempty"`
	Role          string                 `json:"role"`
	Content       string                 `json:"content"`
	Payload       map[string]interface{} `json:"payload"`
	CreatedAt     time.Time              `json:"created_at"`
}

RunMessageResponse is a stable replay record derived from user input and agent message events.

type RunNextAction 

type RunNextAction struct {
	Type            string                 `json:"type"`
	Label           string                 `json:"label"`
	Hint            string                 `json:"hint"`
	Href            string                 `json:"href,omitempty"`
	Method          string                 `json:"method,omitempty"`
	RequiresHuman   bool                   `json:"requires_human,omitempty"`
	ResourceType    string                 `json:"resource_type,omitempty"`
	ResourceID      string                 `json:"resource_id,omitempty"`
	Source          string                 `json:"source,omitempty"`
	AdditionalProps map[string]interface{} `json:"additional_props,omitempty"`
}

RunNextAction is a machine-readable hint for the UI or external clients.

Type is stable enough for clients to branch on. Hint is human-facing.

type RunRequest 

type RunRequest struct {
	AgentID    string                 `json:"agent_id" validate:"required,uuid"`
	Input      map[string]interface{} `json:"input" validate:"required"`
	Metadata   map[string]interface{} `json:"metadata,omitempty"`
	A2AContext *RunA2AContextRequest  `json:"a2a_context,omitempty"`
	// TaskCallback is OpenLinker's canonical callback field. PushNotification,
	// PushNotificationAlias, and PushNotificationConfig are A2A compatibility
	// aliases; taskCallbackConfigFromRunRequest chooses the first non-empty
	// value in this declaration order.
	TaskCallback           *TaskCallbackConfig `json:"task_callback,omitempty"`
	PushNotification       *TaskCallbackConfig `json:"push_notification,omitempty"`
	PushNotificationAlias  *TaskCallbackConfig `json:"pushNotification,omitempty"`
	PushNotificationConfig *TaskCallbackConfig `json:"pushNotificationConfig,omitempty"`
}

RunRequest POST /api/v1/run 请求体。

AgentID 在 service 层 uuid.Parse 校验。 Input 必填，为创作者 endpoint 接收的入参（透传）。 Metadata 可选，平台原样转发给 endpoint，常用于 trace_id / 客户端版本等。

type RunRequirementEvidenceResponse 

type RunRequirementEvidenceResponse struct {
	RunID            string    `json:"run_id"`
	TaskID           string    `json:"task_id"`
	AgentID          string    `json:"agent_id"`
	RequiredSkillIDs []string  `json:"required_skill_ids"`
	RequiredMCPTools []string  `json:"required_mcp_tools"`
	AgentSkillIDs    []string  `json:"agent_skill_ids"`
	MatchedSkillIDs  []string  `json:"matched_skill_ids"`
	MissingSkillIDs  []string  `json:"missing_skill_ids"`
	UsedMCPTools     []string  `json:"used_mcp_tools"`
	MissingMCPTools  []string  `json:"missing_mcp_tools"`
	CoverageStatus   string    `json:"coverage_status"`
	EvidenceSource   string    `json:"evidence_source"`
	CreatedAt        time.Time `json:"created_at"`
}

RunRequirementEvidenceResponse proves which task Skill/MCP requirements were snapshotted onto a run and how the selected Agent covered them.

type RunResponse 

type RunResponse struct {
	RunID               string                 `json:"run_id"`
	AgentID             string                 `json:"agent_id,omitempty"`
	AgentSlug           string                 `json:"agent_slug,omitempty"`
	AgentName           string                 `json:"agent_name,omitempty"`
	AgentConnectionMode string                 `json:"agent_connection_mode,omitempty"`
	Status              string                 `json:"status"`
	Input               map[string]interface{} `json:"input,omitempty"`
	Output              map[string]interface{} `json:"output,omitempty"`
	ErrorCode           string                 `json:"error_code,omitempty"`
	// ErrorMsg keeps the historical Go field name while preserving the public
	// JSON contract as error_message.
	ErrorMsg            string                          `json:"error_message,omitempty"`
	CostCents           int32                           `json:"cost_cents"`
	DurationMs          int32                           `json:"duration_ms"`
	Source              string                          `json:"source,omitempty"`
	ParentRunID         string                          `json:"parent_run_id,omitempty"`
	CallerAgentID       string                          `json:"caller_agent_id,omitempty"`
	BillingMode         string                          `json:"billing_mode,omitempty"`
	A2AContext          *RunA2AContextResponse          `json:"a2a_context,omitempty"`
	TaskCallback        *RunTaskCallbackResponse        `json:"task_callback,omitempty"`
	RequirementEvidence *RunRequirementEvidenceResponse `json:"requirement_evidence,omitempty"`
	EvidenceSummary     *RunEvidenceSummary             `json:"evidence_summary,omitempty"`
	NextAction          *RunNextAction                  `json:"next_action,omitempty"`
}

RunResponse POST /api/v1/run 同步响应体，或 POST /api/v1/runs 异步启动响应体。

Status: 'running' / 'success' / 'failed' / 'timeout'。 失败 / 超时 时 Output 为空、ErrorCode + ErrorMsg 必填，CostCents=0（已退款）。 Source: 'web' / 'mcp' / 'api'，由 handler 从鉴权方式推导。

type RunTaskCallbackResponse 

type RunTaskCallbackResponse struct {
	ID                  string   `json:"id"`
	RunID               string   `json:"run_id"`
	TargetURL           string   `json:"target_url"`
	EventTypes          []string `json:"event_types"`
	AuthScheme          string   `json:"auth_scheme,omitempty"`
	Status              string   `json:"status"`
	ConsecutiveFailures int32    `json:"consecutive_failures"`
	Secret              string   `json:"secret,omitempty"`
	CreatedAt           string   `json:"created_at"`
	UpdatedAt           string   `json:"updated_at"`
}

RunTaskCallbackResponse describes a caller-owned task callback created while starting or delegating a run. Secret is only populated on creation.

type RuntimePullClaimOptions 

type RuntimePullClaimOptions struct {
	Wait time.Duration
}

RuntimePullClaimOptions lets newer workers avoid tight empty polling while keeping the existing GET /claim behavior compatible for older workers.

type RuntimePullResultRequest 

type RuntimePullResultRequest struct {
	Status     string                 `json:"status" validate:"required,oneof=success failed timeout"`
	Output     map[string]interface{} `json:"output,omitempty"`
	Events     []AgentEvent           `json:"events,omitempty"`
	Error      *AgentError            `json:"error,omitempty"`
	DurationMs int32                  `json:"duration_ms,omitempty"`
}

RuntimePullResultRequest 是 runtime_pull Agent 执行完任务后回传的结果。

Status 支持 success / failed / timeout；success 必须带 output，失败建议带 error。

type RuntimePullRunResponse 

type RuntimePullRunResponse struct {
	RunID          string                 `json:"run_id"`
	AgentID        string                 `json:"agent_id"`
	Input          map[string]interface{} `json:"input"`
	Metadata       map[string]interface{} `json:"metadata,omitempty"`
	Source         string                 `json:"source"`
	ResultEndpoint string                 `json:"result_endpoint"`
	ResultMethod   string                 `json:"result_method"`
	ResultRequired bool                   `json:"result_required"`
	A2A            *AgentA2AContext       `json:"a2a,omitempty"`
	Conversation   *ConversationContext   `json:"conversation,omitempty"`
}

RuntimePullRunResponse 是内网 / IPv4 / NAT Agent 主动拉任务时拿到的 payload。

type RuntimePullRunTimeoutConfig 

type RuntimePullRunTimeoutConfig struct {
	DispatchTimeout time.Duration
	ResultTimeout   time.Duration
	BatchSize       int32
}

RuntimePullRunTimeoutConfig controls how long runtime_pull runs may stay pending or claimed before the platform converts them to a terminal timeout.

type RuntimePullRunWorkerConfig 

type RuntimePullRunWorkerConfig struct {
	Interval        time.Duration
	DispatchTimeout time.Duration
	ResultTimeout   time.Duration
	BatchSize       int32
}

type RuntimeWSClientMessage 

type RuntimeWSClientMessage struct {
	Type       string                 `json:"type"`
	ID         string                 `json:"id,omitempty"`
	RunID      string                 `json:"run_id,omitempty"`
	EventType  string                 `json:"event_type,omitempty"`
	Payload    map[string]interface{} `json:"payload,omitempty"`
	Status     string                 `json:"status,omitempty"`
	Output     map[string]interface{} `json:"output,omitempty"`
	Events     []AgentEvent           `json:"events,omitempty"`
	Error      *AgentError            `json:"error,omitempty"`
	DurationMs int32                  `json:"duration_ms,omitempty"`
}

RuntimeWSClientMessage is sent by a NAT/private Agent over /api/v1/agent-runtime/ws. The same runtime token and run state checks used by runtime_pull are reused for result and event writes.

type RuntimeWSServerMessage 

type RuntimeWSServerMessage struct {
	Type              string                  `json:"type"`
	ID                string                  `json:"id,omitempty"`
	RunID             string                  `json:"run_id,omitempty"`
	AgentID           string                  `json:"agent_id,omitempty"`
	Input             map[string]interface{}  `json:"input,omitempty"`
	Metadata          map[string]interface{}  `json:"metadata,omitempty"`
	Source            string                  `json:"source,omitempty"`
	ResultEndpoint    string                  `json:"result_endpoint,omitempty"`
	ResultMethod      string                  `json:"result_method,omitempty"`
	ResultRequired    bool                    `json:"result_required,omitempty"`
	A2A               *AgentA2AContext        `json:"a2a,omitempty"`
	Conversation      *ConversationContext    `json:"conversation,omitempty"`
	Status            string                  `json:"status,omitempty"`
	Result            *RunResponse            `json:"result,omitempty"`
	Event             *RunEventResponse       `json:"event,omitempty"`
	Heartbeat         *AgentHeartbeatResponse `json:"heartbeat,omitempty"`
	Error             *AgentError             `json:"error,omitempty"`
	RetryAfterSeconds int32                   `json:"retry_after_seconds,omitempty"`
}

RuntimeWSServerMessage is sent by OpenLinker over the Agent WebSocket. run.assigned is intentionally flattened so simple JS workers can consume it without understanding nested protocol objects.

type Service 

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

Service 调用 Agent，并维护 core-owned run state, events, artifacts and availability.

关键时序约束（见 docs/13 模块 4 / docs/10 章四）：

  1. 事务 A：INSERT runs(status=running)
  2. 事务外：HTTP POST 创作者 endpoint（60s 超时）
  3. 事务 B：成功 → MarkRunSuccess；失败 → MarkRunFailed 成功统计、availability、artifact 和事件写入不得把已完成的 run 回滚。
  4. 事务外：异步触发 webhook 投递（不阻塞响应）

HTTP 调用必须在事务外，否则会长时间占用数据库事务。

func NewService 

func NewService(pool *pgxpool.Pool, cfg *config.Config) *Service

NewService 构造 Service。HTTP client timeout 取自 cfg.RunTimeoutSeconds（默认 60s）。

func (*Service) CancelRun 

func (s *Service) CancelRun(ctx context.Context, userID, runID uuid.UUID) (*RunResponse, error)

CancelRun marks a running run as canceled. Only the run owner can cancel it.

func (*Service) ClaimRuntimePullRun 

func (s *Service) ClaimRuntimePullRun(ctx context.Context, plaintextToken string, opts ...RuntimePullClaimOptions) (*RuntimePullRunResponse, error)

ClaimRuntimePullRun lets a private / IPv4 Agent actively pull the next pending run. runtime_ws Agents may use this as the fallback path after WebSocket loss.

func (*Service) CompleteRuntimePullRun 

func (s *Service) CompleteRuntimePullRun(ctx context.Context, plaintextToken string, runID uuid.UUID, req *RuntimePullResultRequest) (*RunResponse, error)

CompleteRuntimePullRun accepts the result of a run previously claimed by the same access token.

func (*Service) DryRun 

func (s *Service) DryRun(
	ctx context.Context,
	agent *db.Agent,
	input map[string]interface{},
) (map[string]interface{}, string)

DryRun 让创作者侧 endpoint 跑一次「不计费、不写 runs」的探活调用。

用于 Agent 接入流程的 dry-run 步骤：使用给定输入直接命中 endpoint， 返回 endpoint 的输出或错误信息。runID 用随机 UUID 仅为响应头标识， 没有任何 DB 副作用。

返回 (output, errMsg)：errMsg 为空字符串时表示成功。

func (*Service) GetRun 

func (s *Service) GetRun(ctx context.Context, userID, runID uuid.UUID) (*RunResponse, error)

GetRun 查单条调用详情；仅 owner 可看。

func (*Service) HeartbeatAgent 

func (s *Service) HeartbeatAgent(ctx context.Context, plaintextToken string) (*AgentHeartbeatResponse, error)

HeartbeatAgent lets an Agent proactively mark its bound access-token owner alive.

func (*Service) ListRunArtifacts 

func (s *Service) ListRunArtifacts(ctx context.Context, userID, runID uuid.UUID) ([]RunArtifactResponse, error)

ListRunArtifacts returns persisted artifacts for a run. Only the run owner can read them.

func (*Service) ListRunEvents 

func (s *Service) ListRunEvents(ctx context.Context, userID, runID uuid.UUID, afterSequence, limit int32) ([]RunEventResponse, error)

ListRunEvents 查询单个 run 的事件流；仅 owner 可看。

func (*Service) ListRunMessages 

func (s *Service) ListRunMessages(ctx context.Context, userID, runID uuid.UUID) ([]RunMessageResponse, error)

ListRunMessages returns stable message replay records for a run.

func (*Service) ReportRunEvent 

func (s *Service) ReportRunEvent(ctx context.Context, runID uuid.UUID, token string, req *ReportRunEventRequest) (*RunEventResponse, error)

ReportRunEvent 允许 Agent endpoint 用 endpoint token 上报当前 run 的中间事件。

func (*Service) ReportRuntimeTokenRunEvent 

func (s *Service) ReportRuntimeTokenRunEvent(ctx context.Context, plaintextToken string, runID uuid.UUID, req *ReportRunEventRequest) (*RunEventResponse, error)

ReportRuntimeTokenRunEvent lets runtime_ws workers stream progress over the same authenticated channel used for assignment and result delivery.

func (*Service) Run 

func (s *Service) Run(ctx context.Context, userID uuid.UUID, req *RunRequest, source string) (*RunResponse, error)

Run 调用 Agent。

流程见 Service 注释。Core 不执行商业结算，因此 run 免费。

source 标记调用来源：'web' / 'mcp' / 'api'，写入 runs.source 以便 /usage 分类显示。 传空字符串时默认 'web'，便于旧调用方零修改。

func (*Service) RunDelegated 

func (s *Service) RunDelegated(ctx context.Context, userID uuid.UUID, delegation Delegation, req *RunRequest) (*RunResponse, error)

RunDelegated lets an authenticated Agent call another Agent through the platform. Delegated runs are free until explicit user-approved billing exists.

func (*Service) ServeRuntimeWebSocket 

func (s *Service) ServeRuntimeWebSocket(w http.ResponseWriter, r *http.Request, plaintextToken string) error

ServeRuntimeWebSocket accepts an Agent-owned outbound WebSocket and uses it as a low-latency assignment channel. Runtime state remains DB-backed; the existing long-poll claim/result endpoints remain the fallback path.

func (*Service) SetDeliveryEnqueuer 

func (s *Service) SetDeliveryEnqueuer(d DeliveryEnqueuer)

SetDeliveryEnqueuer 注入用户侧投递触发器（main.go 启动时调用）。

func (*Service) SetTaskCallbackEnqueuer 

func (s *Service) SetTaskCallbackEnqueuer(w TaskCallbackEnqueuer)

SetTaskCallbackEnqueuer 注入 task callback 触发器。

func (*Service) SetWebhookEnqueuer 

func (s *Service) SetWebhookEnqueuer(w WebhookEnqueuer)

SetWebhookEnqueuer 注入 webhook 触发器（main.go 启动时调用）。

用 setter 而非 NewService 参数，避免 runtime ↔ webhook 循环依赖 （webhook 内部要 import runtime 也不行；用接口隔离）。

func (*Service) StartRun 

func (s *Service) StartRun(ctx context.Context, userID uuid.UUID, req *RunRequest, source string) (*RunResponse, error)

StartRun 创建 running run 并在后台执行；调用方可用 GetRun/ListRunEvents/SSE 查询进度。

func (*Service) TimeoutStaleEndpointRuns 

func (s *Service) TimeoutStaleEndpointRuns(ctx context.Context, cfg EndpointRunTimeoutConfig) (int32, error)

TimeoutStaleEndpointRuns converts abandoned direct_http / mcp_server runs into timeout terminal states. Normal endpoint calls are completed by the in-process goroutine; this worker is only a crash / restart / DB outage recovery net for runs that have exceeded the configured stale window.

func (*Service) TimeoutStaleRuntimePullRuns 

func (s *Service) TimeoutStaleRuntimePullRuns(ctx context.Context, cfg RuntimePullRunTimeoutConfig) (int32, error)

TimeoutStaleRuntimePullRuns converts abandoned queued runtime runs into timeout terminal states so users and upstream callers never wait forever.

func (*Service) ValidateRuntimeToken 

func (s *Service) ValidateRuntimeToken(ctx context.Context, plaintext string, acceptedScopes ...string) error

type TaskCallbackAuthentication 

type TaskCallbackAuthentication struct {
	Scheme      string `json:"scheme,omitempty"`
	Credentials string `json:"credentials,omitempty"`
}

type TaskCallbackConfig 

type TaskCallbackConfig struct {
	URL             string                      `json:"url,omitempty"`
	Token           string                      `json:"token,omitempty"`
	Secret          string                      `json:"secret,omitempty"`
	Authentication  *TaskCallbackAuthentication `json:"authentication,omitempty"`
	Metadata        map[string]interface{}      `json:"metadata,omitempty"`
	EventTypes      []string                    `json:"eventTypes,omitempty"`
	EventTypesAlias []string                    `json:"event_types,omitempty"`
}

type TaskCallbackEnqueuer 

type TaskCallbackEnqueuer interface {
	EnqueueRunEvent(ctx context.Context, event db.RunEvent) error
}

TaskCallbackEnqueuer 触发 task callback，payload 来自 run_events。

type WebhookEnqueuer 

type WebhookEnqueuer interface {
	EnqueueDelivery(ctx context.Context, run *db.Run, agentSlug string, output map[string]interface{}) error
}

WebhookEnqueuer 触发 run 完成后向创作者推送 webhook。

用接口注入避免 runtime → webhook 的硬依赖（webhook 本身依赖 db.Run）。 实现见 internal/webhook.Service.EnqueueDelivery。

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.