Files
Agent-System/internal/workflow/runtime/orchestrator.go
T
weichenandCursor 265f09c1f2 fix(server): cache health probes, surface SSE marshal errors, fail closed on missing duties
Health checks reuse a two-second snapshot and bound the schema query. Task SSE writes an error frame when an event cannot be encoded. ResolveDocTask errors when the frozen plan has no duty row, while explicit none-duties stay zero. Drop the unwired semantic-summary replace env, record the intentional dev versus release model split, document the test/main cutover path, and archive pre-goose SQL.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-19 09:16:18 +00:00

109 lines
2.9 KiB
Go

package runtime
import (
"context"
"fmt"
"chat-agent-system/internal/projectinstance"
"chat-agent-system/internal/workflowdef"
)
// TaskResolution 从冻结实例解析的运行时任务绑定。
type TaskResolution struct {
ProjectID int64
WorkflowInstID int64
StageKey string
TaskKey string
DocTaskInstID int64
ExecutorKey string
QualityGateKey string
TemplateAsset string
TemplateVersion int
AuthorUserID int64
ReviewerUserID int64
ApproverUserID int64
}
// Orchestrator 只读冻结快照编排(禁止 flatten 旁路)。
type Orchestrator struct {
instSvc *projectinstance.Service
}
func NewOrchestrator(instSvc *projectinstance.Service) *Orchestrator {
return &Orchestrator{instSvc: instSvc}
}
// ResolveDocTask 由 doc_task_instance_id 解析 executor/模板/三权。
func (o *Orchestrator) ResolveDocTask(ctx context.Context, projectID, docTaskInstID int64) (*TaskResolution, error) {
if o == nil || o.instSvc == nil || !o.instSvc.Available() {
return nil, fmt.Errorf("orchestrator unavailable")
}
detail, version, err := o.instSvc.GetLiveInstanceDetail(ctx, projectID)
if err != nil {
return nil, err
}
plan := detail.Instance.ConfigSnapshot
for _, st := range detail.Stages {
for _, dt := range st.DocTasks {
if dt.ID != docTaskInstID {
continue
}
res := &TaskResolution{
ProjectID: projectID,
WorkflowInstID: detail.Instance.ID,
StageKey: st.StageKey,
TaskKey: dt.TaskKey,
DocTaskInstID: dt.ID,
}
if def := lookupTaskDef(&version.Definition, st.StageKey, dt.TaskKey); def != nil {
res.ExecutorKey = def.ExecutorKey
res.QualityGateKey = def.QualityGateKey
if def.Template != nil {
res.TemplateAsset = def.Template.AssetKey
res.TemplateVersion = def.Template.Version
} else {
res.TemplateAsset = def.TemplateAssetKey
res.TemplateVersion = def.TemplateVersion
}
}
matchedDuty := false
for _, stPlan := range plan.Stages {
if stPlan.StageKey != st.StageKey {
continue
}
for _, dtPlan := range stPlan.DocTasks {
if dtPlan.TaskKey != dt.TaskKey {
continue
}
res.AuthorUserID = dtPlan.Duties.AuthorUserID
res.ReviewerUserID = dtPlan.Duties.ReviewerUserID
res.ApproverUserID = dtPlan.Duties.ApproverUserID
matchedDuty = true
}
}
if !matchedDuty {
return nil, fmt.Errorf("orchestrator: frozen plan has no duties for stage %s task %s", st.StageKey, dt.TaskKey)
}
return res, nil
}
}
return nil, fmt.Errorf("doc task instance not in live snapshot")
}
func lookupTaskDef(def *workflowdef.VersionDefinition, stageKey, taskKey string) *workflowdef.DocTaskDef {
if def == nil {
return nil
}
for _, st := range def.Stages {
if st.Key != stageKey {
continue
}
for i := range st.DocTasks {
if st.DocTasks[i].Key == taskKey {
return &st.DocTasks[i]
}
}
}
return nil
}