diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 4e715c16..3cfb4169 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -253,6 +253,9 @@ imports: controlcenter/docs: - controlcenter/embed - controlcenter/store + controlcenter/importer: + - controlcenter/pipeline + - controlcenter/store controlcenter/pipeline: [] controlcenter/store: [] controlcenter/tasks: diff --git a/cmd/control-center/agentcli.go b/cmd/control-center/agentcli.go new file mode 100644 index 00000000..4551e12e --- /dev/null +++ b/cmd/control-center/agentcli.go @@ -0,0 +1,401 @@ +package main + +import ( + "bytes" + "encoding/json" + "flag" + "fmt" + "io" + "net/http" + "os" + "strings" + "time" +) + +// Thin agent-side HTTP client over the Control Center agent API. Output is +// compact JSON on stdout; failures print the server's error field to stderr +// and exit non-zero. + +type agentClient struct { + base string + token string + hc *http.Client +} + +func newAgentClient(server, token string) (*agentClient, error) { + if token == "" { + return nil, fmt.Errorf("agent token required: pass -token or set CC_AGENT_TOKEN") + } + return &agentClient{ + base: strings.TrimRight(server, "/"), + token: token, + hc: &http.Client{Timeout: 30 * time.Second}, + }, nil +} + +// do performs one request and returns the decoded body. Server error +// responses are unwrapped into an error carrying the server's error field. +func (c *agentClient) do(method, path string, body any) (json.RawMessage, error) { + var rdr io.Reader + if body != nil { + raw, err := json.Marshal(body) + if err != nil { + return nil, err + } + rdr = bytes.NewReader(raw) + } + req, err := http.NewRequest(method, c.base+path, rdr) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+c.token) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + resp, err := c.hc.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + raw, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if resp.StatusCode == http.StatusNoContent { + return nil, nil + } + if resp.StatusCode >= 400 { + var e struct { + Error string `json:"error"` + } + if json.Unmarshal(raw, &e) == nil && e.Error != "" { + return nil, fmt.Errorf("%s %s: %s (%d)", method, path, e.Error, resp.StatusCode) + } + return nil, fmt.Errorf("%s %s: HTTP %d: %s", method, path, resp.StatusCode, strings.TrimSpace(string(raw))) + } + return json.RawMessage(raw), nil +} + +func printJSON(raw json.RawMessage) { + if len(raw) == 0 { + fmt.Println("null") + return + } + // Re-encode compactly so output is stable regardless of server framing. + var v any + if err := json.Unmarshal(raw, &v); err != nil { + fmt.Println(string(raw)) + return + } + out, _ := json.Marshal(v) + fmt.Println(string(out)) +} + +func envDefault(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +// commonFlags registers -server and -token on fs and returns a client +// factory. +func commonFlags(fs *flag.FlagSet) func() (*agentClient, error) { + server := fs.String("server", envDefault("CC_SERVER", "http://127.0.0.1:8790"), "control-center base URL") + token := fs.String("token", envDefault("CC_AGENT_TOKEN", ""), "agent token (or CC_AGENT_TOKEN)") + return func() (*agentClient, error) { return newAgentClient(*server, *token) } +} + +func readContent(file, content string) (string, error) { + if file != "" { + raw, err := os.ReadFile(file) + if err != nil { + return "", err + } + return string(raw), nil + } + if content == "" { + return "", fmt.Errorf("one of -file or -content is required") + } + return content, nil +} + +func cmdAgentNext(args []string) error { + fs := flag.NewFlagSet("next", flag.ContinueOnError) + mk := commonFlags(fs) + if err := fs.Parse(args); err != nil { + return err + } + c, err := mk() + if err != nil { + return err + } + raw, err := c.do("GET", "/api/v1/agent/tasks/next", nil) + if err != nil { + return err + } + if len(raw) == 0 { + fmt.Println(`{"task":null}`) + return nil + } + var t struct { + ID string `json:"id"` + Slug string `json:"slug"` + Title string `json:"title"` + Stage string `json:"stage"` + } + if err := json.Unmarshal(raw, &t); err != nil { + return err + } + out, _ := json.Marshal(map[string]string{ + "id": t.ID, "slug": t.Slug, "title": t.Title, "stage": t.Stage, + }) + fmt.Println(string(out)) + return nil +} + +func cmdAgentInstruction(args []string) error { + fs := flag.NewFlagSet("instruction", flag.ContinueOnError) + mk := commonFlags(fs) + task := fs.String("task", "", "task id (uuid; use `next` to discover)") + asMD := fs.Bool("md", false, "render a readable summary instead of raw JSON") + if err := fs.Parse(args); err != nil { + return err + } + if *task == "" { + return fmt.Errorf("-task is required") + } + c, err := mk() + if err != nil { + return err + } + raw, err := c.do("GET", "/api/v1/agent/tasks/"+*task+"/instruction", nil) + if err != nil { + return err + } + if !*asMD { + printJSON(raw) + return nil + } + var pkg struct { + Task struct { + Slug string `json:"slug"` + Title string `json:"title"` + InstructionMD string `json:"instruction_md"` + } `json:"task"` + Stage string `json:"stage"` + StageState string `json:"stage_state"` + Deliverable string `json:"deliverable"` + DecisionRules string `json:"decision_rules"` + } + if err := json.Unmarshal(raw, &pkg); err != nil { + return err + } + fmt.Printf("# %s (%s)\nstage: %s [%s]\n\n## 交付要求\n%s\n\n## 决策规则\n%s\n\n## 任务说明\n%s\n", + pkg.Task.Title, pkg.Task.Slug, pkg.Stage, pkg.StageState, + pkg.Deliverable, pkg.DecisionRules, pkg.Task.InstructionMD) + return nil +} + +// simpleTaskAction covers claim / heartbeat / release: POST with no body. +func simpleTaskAction(name string) func([]string) error { + return func(args []string) error { + fs := flag.NewFlagSet(name, flag.ContinueOnError) + mk := commonFlags(fs) + task := fs.String("task", "", "task id (uuid)") + if err := fs.Parse(args); err != nil { + return err + } + if *task == "" { + return fmt.Errorf("-task is required") + } + c, err := mk() + if err != nil { + return err + } + raw, err := c.do("POST", "/api/v1/agent/tasks/"+*task+"/"+name, nil) + if err != nil { + return err + } + printJSON(raw) + return nil + } +} + +func cmdAgentEvent(args []string) error { + fs := flag.NewFlagSet("event", flag.ContinueOnError) + mk := commonFlags(fs) + task := fs.String("task", "", "task id (uuid)") + phase := fs.String("phase", "", "event phase: start|plan|progress|blocked|report|done|advance-request|decision|system") + message := fs.String("message", "", "event message") + payload := fs.String("payload", "", "optional JSON payload") + if err := fs.Parse(args); err != nil { + return err + } + if *task == "" || *phase == "" { + return fmt.Errorf("-task and -phase are required") + } + body := map[string]any{"phase": *phase, "message": *message} + if *payload != "" { + var p any + if err := json.Unmarshal([]byte(*payload), &p); err != nil { + return fmt.Errorf("-payload must be valid JSON: %w", err) + } + body["payload"] = json.RawMessage(*payload) + } + c, err := mk() + if err != nil { + return err + } + raw, err := c.do("POST", "/api/v1/agent/tasks/"+*task+"/events", body) + if err != nil { + return err + } + printJSON(raw) + return nil +} + +func cmdAgentArtifact(args []string) error { + fs := flag.NewFlagSet("artifact", flag.ContinueOnError) + mk := commonFlags(fs) + task := fs.String("task", "", "task id (uuid)") + kind := fs.String("kind", "note", "artifact kind, e.g. proposal|analysis|design|implementation|test-report|evaluation") + title := fs.String("title", "", "artifact title") + file := fs.String("file", "", "read content_md from file") + content := fs.String("content", "", "inline content_md") + if err := fs.Parse(args); err != nil { + return err + } + if *task == "" { + return fmt.Errorf("-task is required") + } + md, err := readContent(*file, *content) + if err != nil { + return err + } + c, err := mk() + if err != nil { + return err + } + raw, err := c.do("POST", "/api/v1/agent/tasks/"+*task+"/artifacts", map[string]string{ + "kind": *kind, "title": *title, "content_md": md, + }) + if err != nil { + return err + } + printJSON(raw) + return nil +} + +func cmdAgentAdvanceRequest(args []string) error { + fs := flag.NewFlagSet("advance-request", flag.ContinueOnError) + mk := commonFlags(fs) + task := fs.String("task", "", "task id (uuid)") + summary := fs.String("summary", "", "handoff summary for human review") + artifacts := fs.String("artifacts", "", "comma-separated artifact ids") + if err := fs.Parse(args); err != nil { + return err + } + if *task == "" { + return fmt.Errorf("-task is required") + } + ids := []string{} + if *artifacts != "" { + for _, id := range strings.Split(*artifacts, ",") { + if id = strings.TrimSpace(id); id != "" { + ids = append(ids, id) + } + } + } + c, err := mk() + if err != nil { + return err + } + raw, err := c.do("POST", "/api/v1/agent/tasks/"+*task+"/advance-request", map[string]any{ + "summary": *summary, "artifact_ids": ids, + }) + if err != nil { + return err + } + printJSON(raw) + return nil +} + +// suggestions collects repeated -suggestion "选项||理由||影响" flags. +type suggestions []map[string]string + +func (s *suggestions) String() string { return fmt.Sprint([]map[string]string(*s)) } +func (s *suggestions) Set(v string) error { + parts := strings.SplitN(v, "||", 3) + if len(parts) != 3 { + return fmt.Errorf("suggestion must be \"选项||理由||影响\"") + } + *s = append(*s, map[string]string{ + "option": strings.TrimSpace(parts[0]), + "rationale": strings.TrimSpace(parts[1]), + "impact": strings.TrimSpace(parts[2]), + }) + return nil +} + +func cmdAgentDecision(args []string) error { + fs := flag.NewFlagSet("decision", flag.ContinueOnError) + mk := commonFlags(fs) + task := fs.String("task", "", "task id (uuid)") + question := fs.String("question", "", "question for the human (required)") + contextMD := fs.String("context", "", "markdown context") + var suggs suggestions + fs.Var(&suggs, "suggestion", `"选项||理由||影响"; repeatable, at least one required`) + if err := fs.Parse(args); err != nil { + return err + } + if *task == "" || *question == "" { + return fmt.Errorf("-task and -question are required") + } + if len(suggs) == 0 { + return fmt.Errorf("at least one -suggestion is required") + } + c, err := mk() + if err != nil { + return err + } + raw, err := c.do("POST", "/api/v1/agent/tasks/"+*task+"/decisions", map[string]any{ + "question": *question, "context_md": *contextMD, "suggestions": suggs, + }) + if err != nil { + return err + } + printJSON(raw) + return nil +} + +func cmdAgentReport(args []string) error { + fs := flag.NewFlagSet("report", flag.ContinueOnError) + mk := commonFlags(fs) + task := fs.String("task", "", "task id (uuid)") + kind := fs.String("kind", "", "result|test-report|handoff") + file := fs.String("file", "", "read content_md from file") + content := fs.String("content", "", "inline content_md") + if err := fs.Parse(args); err != nil { + return err + } + if *task == "" || *kind == "" { + return fmt.Errorf("-task and -kind are required") + } + md, err := readContent(*file, *content) + if err != nil { + return err + } + c, err := mk() + if err != nil { + return err + } + raw, err := c.do("POST", "/api/v1/agent/reports", map[string]string{ + "task_id": *task, "kind": *kind, "content_md": md, + }) + if err != nil { + return err + } + printJSON(raw) + return nil +} diff --git a/cmd/control-center/main.go b/cmd/control-center/main.go index e9e3ce75..81068253 100644 --- a/cmd/control-center/main.go +++ b/cmd/control-center/main.go @@ -1,5 +1,7 @@ // Control Center: personal project control plane, independent of the product -// system. Subcommands: serve, migrate, register-agent. +// system. Subcommands: serve, migrate, register-agent, import-docs, plus the +// agent CLI (next / instruction / claim / heartbeat / release / event / +// artifact / advance-request / decision / report). package main import ( @@ -18,6 +20,7 @@ import ( "chat-agent-system/internal/controlcenter/config" "chat-agent-system/internal/controlcenter/docs" "chat-agent-system/internal/controlcenter/embed" + "chat-agent-system/internal/controlcenter/importer" "chat-agent-system/internal/controlcenter/server" "chat-agent-system/internal/controlcenter/store" ) @@ -35,6 +38,28 @@ func main() { err = cmdMigrate(os.Args[2:]) case "register-agent": err = cmdRegisterAgent(os.Args[2:]) + case "import-docs": + err = cmdImportDocs(os.Args[2:]) + case "next": + err = cmdAgentNext(os.Args[2:]) + case "instruction": + err = cmdAgentInstruction(os.Args[2:]) + case "claim": + err = simpleTaskAction("claim")(os.Args[2:]) + case "heartbeat": + err = simpleTaskAction("heartbeat")(os.Args[2:]) + case "release": + err = simpleTaskAction("release")(os.Args[2:]) + case "event": + err = cmdAgentEvent(os.Args[2:]) + case "artifact": + err = cmdAgentArtifact(os.Args[2:]) + case "advance-request": + err = cmdAgentAdvanceRequest(os.Args[2:]) + case "decision": + err = cmdAgentDecision(os.Args[2:]) + case "report": + err = cmdAgentReport(os.Args[2:]) default: usage() os.Exit(2) @@ -51,7 +76,19 @@ func usage() { usage: control-center serve run HTTP server (CC_HTTP_ADDR, default 127.0.0.1:8790) control-center migrate apply goose migrations only - control-center register-agent -name X [-kind K] [-token T] create/rotate an agent token`) + control-center register-agent -name X [-kind K] [-token T] create/rotate an agent token + control-center import-docs import .doc task docs into cc_tasks (also runs periodically in serve) + +agent CLI (common flags: -server, default $CC_SERVER or http://127.0.0.1:8790; + -token, default $CC_AGENT_TOKEN; -task takes a task uuid — use 'next' to discover ids): + next print next claimable task (slug/title/id/stage) + instruction -task ID [-md] print the instruction package (JSON, or readable with -md) + claim|heartbeat|release -task ID lease lifecycle + event -task ID -phase P -message M [-payload ''] + artifact -task ID -kind K -title T [-file F | -content MD] + advance-request -task ID -summary S [-artifacts id1,id2] + decision -task ID -question Q -context MD -suggestion "选项||理由||影响" [-suggestion ...] + report -task ID -kind result|test-report|handoff [-file F | -content MD]`) } func cmdServe(args []string) error { @@ -95,6 +132,7 @@ func cmdServe(args []string) error { syncCtx, stopSync := context.WithCancel(ctx) defer stopSync() syncer.StartPeriodic(syncCtx, time.Duration(cfg.DocSyncSecond)*time.Second) + importer.New(st, cfg.RepoRoot).StartPeriodic(syncCtx, time.Duration(cfg.DocSyncSecond)*time.Second) reapCtx, stopReaper := context.WithCancel(ctx) defer stopReaper() @@ -131,6 +169,29 @@ func cmdMigrate(args []string) error { return store.Migrate(context.Background(), cfg.DBDSN, cfg.MigrationsDir) } +func cmdImportDocs(args []string) error { + fs := flag.NewFlagSet("import-docs", flag.ContinueOnError) + if err := fs.Parse(args); err != nil { + return err + } + cfg, err := config.Load() + if err != nil { + return err + } + ctx := context.Background() + st, err := store.Connect(ctx, cfg.DBDSN) + if err != nil { + return err + } + defer st.Close() + stats, err := importer.New(st, cfg.RepoRoot).Import(ctx) + if err != nil { + return err + } + fmt.Printf("import-docs: added=%d updated=%d skipped=%d\n", stats.Added, stats.Updated, stats.Skipped) + return nil +} + func cmdRegisterAgent(args []string) error { fs := flag.NewFlagSet("register-agent", flag.ContinueOnError) name := fs.String("name", "", "agent name (required)") diff --git a/internal/controlcenter/importer/importer.go b/internal/controlcenter/importer/importer.go new file mode 100644 index 00000000..4d8a2f9e --- /dev/null +++ b/internal/controlcenter/importer/importer.go @@ -0,0 +1,356 @@ +// Package importer ingests existing .doc task documents (ops/short-term, +// ops/long-term, backlog) into cc_tasks, keyed by slug (file name without +// .md). Re-imports are idempotent: unchanged documents are skipped, and +// fields a human edited in the UI (priority, stage, assignment, terminal +// status) are never overwritten. +package importer + +import ( + "context" + "errors" + "fmt" + "io/fs" + "log/slog" + "os" + "path/filepath" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "gopkg.in/yaml.v3" + + "chat-agent-system/internal/controlcenter/pipeline" + "chat-agent-system/internal/controlcenter/store" +) + +// planDirs maps a .doc task-doc directory to its plan slug and horizon. +var planDirs = []struct { + dir string // under .doc/ + slug string + title string + horizon string +}{ + {"ops/short-term", "short-term", "短期任务", "short"}, + {"ops/long-term", "long-term", "长期任务", "long"}, + {"backlog", "backlog", "Backlog", "backlog"}, +} + +type Importer struct { + st *store.Store + repoRoot string +} + +func New(st *store.Store, repoRoot string) *Importer { + return &Importer{st: st, repoRoot: repoRoot} +} + +// Stats summarizes one Import run. +type Stats struct { + Added int `json:"added"` + Updated int `json:"updated"` + Skipped int `json:"skipped"` +} + +// docTask is one parsed task document. +type docTask struct { + slug string + title string + status string // cc_tasks status + stage string + stageState string + owner string + branch string + worktree string + mutex string + raw string // full file content, stored as instruction_md + docPath string // repo-relative slash path + planSlug string +} + +// Import scans the task-doc directories and upserts one cc_tasks row per +// document. See package doc for the overwrite rules. +func (im *Importer) Import(ctx context.Context) (Stats, error) { + var stats Stats + planIDs := map[string]string{} + for _, pd := range planDirs { + id, err := im.ensurePlan(ctx, pd.slug, pd.title, pd.horizon) + if err != nil { + return stats, fmt.Errorf("ensure plan %s: %w", pd.slug, err) + } + planIDs[pd.slug] = id + } + + // The same file name may live in more than one task directory (e.g. a + // doc copied from short-term to long-term). Directory order in planDirs + // decides the winner; later duplicates are skipped so re-runs stay + // idempotent instead of ping-ponging the row. + seen := map[string]string{} + for _, pd := range planDirs { + root := filepath.Join(im.repoRoot, ".doc", filepath.FromSlash(pd.dir)) + err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() || !strings.HasSuffix(d.Name(), ".md") { + return nil + } + dt, err := im.parseDoc(path, pd.slug) + if err != nil { + slog.Warn("doc task import: skip unparseable doc", "path", path, "err", err) + stats.Skipped++ + return nil + } + if prev, dup := seen[dt.slug]; dup { + slog.Warn("doc task import: duplicate slug, keeping first", + "slug", dt.slug, "kept", prev, "skipped", dt.docPath) + stats.Skipped++ + return nil + } + seen[dt.slug] = dt.docPath + res, err := im.upsert(ctx, dt, planIDs[pd.slug]) + if err != nil { + return fmt.Errorf("upsert %s: %w", dt.docPath, err) + } + switch res { + case "added": + stats.Added++ + case "updated": + stats.Updated++ + default: + stats.Skipped++ + } + return nil + }) + if err != nil { + if os.IsNotExist(err) { + continue + } + return stats, err + } + } + return stats, nil +} + +// StartPeriodic runs Import immediately, then every interval until ctx is +// cancelled (same ticker pattern as docs.Syncer.StartPeriodic). +func (im *Importer) StartPeriodic(ctx context.Context, interval time.Duration) { + go func() { + im.runOnce(ctx) + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + im.runOnce(ctx) + } + } + }() +} + +func (im *Importer) runOnce(ctx context.Context) { + stats, err := im.Import(ctx) + if err != nil { + slog.Error("doc task import", "err", err) + return + } + slog.Info("doc task import done", + "added", stats.Added, "updated", stats.Updated, "skipped", stats.Skipped) +} + +func (im *Importer) ensurePlan(ctx context.Context, slug, title, horizon string) (string, error) { + var id string + err := im.st.Pool.QueryRow(ctx, + `INSERT INTO cc_plans (slug, title, horizon) VALUES ($1,$2,$3) + ON CONFLICT (slug) DO NOTHING RETURNING id`, slug, title, horizon).Scan(&id) + if errors.Is(err, pgx.ErrNoRows) { + err = im.st.Pool.QueryRow(ctx, `SELECT id FROM cc_plans WHERE slug = $1`, slug).Scan(&id) + } + return id, err +} + +// parseDoc reads one markdown file into a docTask. +func (im *Importer) parseDoc(path, planSlug string) (*docTask, error) { + rawBytes, err := os.ReadFile(path) + if err != nil { + return nil, err + } + raw := string(rawBytes) + fm, body := splitFrontMatter(raw) + + rel, err := filepath.Rel(im.repoRoot, path) + if err != nil { + return nil, err + } + base := filepath.Base(path) + slug := strings.TrimSuffix(base, ".md") + + status := mapStatus(fmString(fm, "status")) + stage, stageState := inferStage(status) + return &docTask{ + slug: slug, + title: docTitle(fm, body, slug), + status: status, + stage: stage, + stageState: stageState, + owner: fmString(fm, "owner"), + branch: fmString(fm, "branch"), + worktree: fmString(fm, "worktree"), + mutex: fmString(fm, "mutex"), + raw: raw, + docPath: filepath.ToSlash(rel), + planSlug: planSlug, + }, nil +} + +// mapStatus maps a doc front-matter status onto the cc_tasks status space. +// Unknown or missing statuses fall back to planned. +func mapStatus(s string) string { + switch strings.TrimSpace(s) { + case pipeline.StatusPlanned: + return pipeline.StatusPlanned + case pipeline.StatusInProgress: + return pipeline.StatusInProgress + case "blocked": + return pipeline.StatusBlockedDecision + case pipeline.StatusDone: + return pipeline.StatusDone + case pipeline.StatusCancelled: + return pipeline.StatusCancelled + default: + return pipeline.StatusPlanned + } +} + +// inferStage derives the initial stage/stage_state for a newly imported +// task. Stage is only set on insert; later syncs never move a task's stage. +func inferStage(status string) (stage, stageState string) { + switch status { + case pipeline.StatusDone: + return pipeline.StageConfirmation, pipeline.StateApproved + case pipeline.StatusInProgress: + return pipeline.StageImplementation, pipeline.StateInProgress + case pipeline.StatusBlockedDecision: + return pipeline.StageImplementation, pipeline.StateChangesRequested + default: // planned, cancelled + return pipeline.StageProposal, pipeline.StateIdle + } +} + +// upsert inserts the task or updates the doc-owned fields of an existing +// one. Returns "added", "updated" or "skipped". +// +// Overwrite rules for existing tasks: +// - title/owner/branch/worktree/mutex/instruction_md/doc_path follow the doc +// - plan_id is backfilled only when the task has none +// - status follows the doc only while the task is planned or in_progress; +// done/cancelled/blocked_pending_decision are never moved by the importer +// - stage/stage_state/priority/assignment are never touched after insert +func (im *Importer) upsert(ctx context.Context, dt *docTask, planID string) (string, error) { + var cur struct { + id string + title string + owner string + branch string + worktree string + mutex string + instr string + docPath string + status string + planID *string + } + err := im.st.Pool.QueryRow(ctx, + `SELECT id, title, owner, branch, worktree, mutex, instruction_md, doc_path, status, plan_id + FROM cc_tasks WHERE slug = $1`, dt.slug). + Scan(&cur.id, &cur.title, &cur.owner, &cur.branch, &cur.worktree, &cur.mutex, + &cur.instr, &cur.docPath, &cur.status, &cur.planID) + if errors.Is(err, pgx.ErrNoRows) { + _, err := im.st.Pool.Exec(ctx, + `INSERT INTO cc_tasks (slug, plan_id, title, status, stage, stage_state, + owner, branch, worktree, mutex, instruction_md, doc_path) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)`, + dt.slug, planID, dt.title, dt.status, dt.stage, dt.stageState, + dt.owner, dt.branch, dt.worktree, dt.mutex, dt.raw, dt.docPath) + if err != nil { + return "", err + } + return "added", nil + } + if err != nil { + return "", err + } + + followStatus := cur.status == pipeline.StatusPlanned || cur.status == pipeline.StatusInProgress + newStatus := cur.status + if followStatus { + newStatus = dt.status + } + changed := cur.title != dt.title || cur.owner != dt.owner || cur.branch != dt.branch || + cur.worktree != dt.worktree || cur.mutex != dt.mutex || cur.instr != dt.raw || + cur.docPath != dt.docPath || cur.status != newStatus || cur.planID == nil + if !changed { + return "skipped", nil + } + newPlanID := cur.planID + if newPlanID == nil { + newPlanID = &planID + } + _, err = im.st.Pool.Exec(ctx, + `UPDATE cc_tasks SET title=$2, owner=$3, branch=$4, worktree=$5, mutex=$6, + instruction_md=$7, doc_path=$8, status=$9, plan_id=$10, updated_at=now() + WHERE id = $1`, + cur.id, dt.title, dt.owner, dt.branch, dt.worktree, dt.mutex, + dt.raw, dt.docPath, newStatus, newPlanID) + if err != nil { + return "", err + } + return "updated", nil +} + +// splitFrontMatter extracts a leading --- yaml --- block; returns the parsed +// map (possibly empty) and the remaining body. +func splitFrontMatter(md string) (map[string]any, string) { + fm := map[string]any{} + body := md + if rest, ok := strings.CutPrefix(md, "---\n"); ok { + if end := strings.Index(rest, "\n---"); end >= 0 { + rawFM := rest[:end] + after := rest[end+len("\n---"):] + if after == "" || after[0] == '\n' || after[0] == '\r' { + if err := yaml.Unmarshal([]byte(rawFM), &fm); err != nil { + fm = map[string]any{} + } else { + body = strings.TrimLeft(after, "\r\n") + } + } + } + } + if fm == nil { + fm = map[string]any{} + } + return fm, body +} + +func fmString(fm map[string]any, key string) string { + if v, ok := fm[key].(string); ok { + return strings.TrimSpace(v) + } + return "" +} + +// docTitle prefers front matter title, then the first level-1 heading, then +// the slug. +func docTitle(fm map[string]any, body, slug string) string { + if t := fmString(fm, "title"); t != "" { + return t + } + for _, line := range strings.Split(body, "\n") { + if h, ok := strings.CutPrefix(strings.TrimSpace(line), "# "); ok { + if t := strings.TrimSpace(h); t != "" { + return t + } + } + } + return slug +} diff --git a/internal/controlcenter/importer/importer_test.go b/internal/controlcenter/importer/importer_test.go new file mode 100644 index 00000000..dcb01e3f --- /dev/null +++ b/internal/controlcenter/importer/importer_test.go @@ -0,0 +1,180 @@ +package importer + +import ( + "fmt" + "os" + "path/filepath" + "runtime" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "chat-agent-system/internal/controlcenter/pipeline" + "chat-agent-system/internal/controlcenter/store" +) + +func TestMapStatus(t *testing.T) { + assert.Equal(t, pipeline.StatusPlanned, mapStatus("planned")) + assert.Equal(t, pipeline.StatusInProgress, mapStatus("in_progress")) + assert.Equal(t, pipeline.StatusBlockedDecision, mapStatus("blocked")) + assert.Equal(t, pipeline.StatusDone, mapStatus("done")) + assert.Equal(t, pipeline.StatusCancelled, mapStatus("cancelled")) + assert.Equal(t, pipeline.StatusPlanned, mapStatus("")) + assert.Equal(t, pipeline.StatusPlanned, mapStatus("weird")) +} + +func TestInferStage(t *testing.T) { + st, ss := inferStage(pipeline.StatusDone) + assert.Equal(t, pipeline.StageConfirmation, st) + assert.Equal(t, pipeline.StateApproved, ss) + + st, ss = inferStage(pipeline.StatusInProgress) + assert.Equal(t, pipeline.StageImplementation, st) + assert.Equal(t, pipeline.StateInProgress, ss) + + st, ss = inferStage(pipeline.StatusBlockedDecision) + assert.Equal(t, pipeline.StageImplementation, st) + assert.Equal(t, pipeline.StateChangesRequested, ss) + + for _, s := range []string{pipeline.StatusPlanned, pipeline.StatusCancelled} { + st, ss = inferStage(s) + assert.Equal(t, pipeline.StageProposal, st) + assert.Equal(t, pipeline.StateIdle, ss) + } +} + +func writeDoc(t *testing.T, repoRoot, rel, content string) { + t.Helper() + full := filepath.Join(repoRoot, filepath.FromSlash(rel)) + require.NoError(t, os.MkdirAll(filepath.Dir(full), 0o755)) + require.NoError(t, os.WriteFile(full, []byte(content), 0o644)) +} + +func TestParseDoc(t *testing.T) { + repoRoot := t.TempDir() + writeDoc(t, repoRoot, ".doc/ops/short-term/2026-09-17-demo-task.md", `--- +title: "示例任务" +status: blocked +owner: kimi +branch: feature/demo +worktree: "-" +mutex: docgov +--- + +# 示例任务 + +## 目标 + +做点什么。 +`) + im := New(nil, repoRoot) + dt, err := im.parseDoc(filepath.Join(repoRoot, ".doc", "ops", "short-term", "2026-09-17-demo-task.md"), "short-term") + require.NoError(t, err) + assert.Equal(t, "2026-09-17-demo-task", dt.slug) + assert.Equal(t, "示例任务", dt.title) + assert.Equal(t, pipeline.StatusBlockedDecision, dt.status) + assert.Equal(t, "kimi", dt.owner) + assert.Equal(t, "feature/demo", dt.branch) + assert.Equal(t, "-", dt.worktree) + assert.Equal(t, "docgov", dt.mutex) + assert.Equal(t, ".doc/ops/short-term/2026-09-17-demo-task.md", dt.docPath) + assert.Contains(t, dt.raw, "## 目标") +} + +func TestParseDocTitleFallback(t *testing.T) { + repoRoot := t.TempDir() + writeDoc(t, repoRoot, ".doc/backlog/no-frontmatter.md", "# 标题来自正文\n\n内容。\n") + im := New(nil, repoRoot) + dt, err := im.parseDoc(filepath.Join(repoRoot, ".doc", "backlog", "no-frontmatter.md"), "backlog") + require.NoError(t, err) + assert.Equal(t, "标题来自正文", dt.title) + assert.Equal(t, pipeline.StatusPlanned, dt.status) + assert.Equal(t, pipeline.StageProposal, dt.stage) + + writeDoc(t, repoRoot, ".doc/backlog/plain.md", "没有任何标题。\n") + dt, err = im.parseDoc(filepath.Join(repoRoot, ".doc", "backlog", "plain.md"), "backlog") + require.NoError(t, err) + assert.Equal(t, "plain", dt.title) +} + +// --- DB integration (TEST_CC_DB_DSN gated) --- + +func newTestStore(t *testing.T) *store.Store { + t.Helper() + dsn := os.Getenv("TEST_CC_DB_DSN") + if dsn == "" { + t.Skip("TEST_CC_DB_DSN not set") + } + _, thisFile, _, _ := runtime.Caller(0) + migrationsDir := filepath.Join(filepath.Dir(thisFile), "..", "..", "..", "deploy", "control-center", "goose") + ctx := t.Context() + require.NoError(t, store.Migrate(ctx, dsn, migrationsDir)) + st, err := store.Connect(ctx, dsn) + require.NoError(t, err) + t.Cleanup(st.Close) + return st +} + +func TestImportIntegration(t *testing.T) { + st := newTestStore(t) + tag := fmt.Sprintf("imp%d", time.Now().UnixNano()) + repoRoot := t.TempDir() + slugDone := tag + "-done" + slugPlanned := tag + "-planned" + + writeDoc(t, repoRoot, ".doc/ops/short-term/"+slugDone+".md", + "---\ntitle: \"已完成任务 "+tag+"\"\nstatus: done\nowner: cursor\nbranch: fix/x\nworktree: \"-\"\nmutex: docgov\n---\n\n# 已完成任务\n\n正文。\n") + writeDoc(t, repoRoot, ".doc/ops/long-term/"+slugPlanned+".md", + "---\ntitle: \"计划中任务 "+tag+"\"\nstatus: planned\n---\n\n# 计划中任务\n\n正文。\n") + + im := New(st, repoRoot) + stats, err := im.Import(t.Context()) + require.NoError(t, err) + assert.Equal(t, 2, stats.Added) + assert.Equal(t, 0, stats.Updated) + + var stage, stageState, status, planSlug, owner string + require.NoError(t, st.Pool.QueryRow(t.Context(), + `SELECT t.stage, t.stage_state, t.status, p.slug, t.owner + FROM cc_tasks t JOIN cc_plans p ON p.id = t.plan_id WHERE t.slug = $1`, slugDone). + Scan(&stage, &stageState, &status, &planSlug, &owner)) + assert.Equal(t, pipeline.StageConfirmation, stage) + assert.Equal(t, pipeline.StateApproved, stageState) + assert.Equal(t, pipeline.StatusDone, status) + assert.Equal(t, "short-term", planSlug) + assert.Equal(t, "cursor", owner) + + require.NoError(t, st.Pool.QueryRow(t.Context(), + `SELECT p.slug FROM cc_tasks t JOIN cc_plans p ON p.id = t.plan_id WHERE t.slug = $1`, slugPlanned). + Scan(&planSlug)) + assert.Equal(t, "long-term", planSlug) + + // Idempotent: second run skips everything. + stats, err = im.Import(t.Context()) + require.NoError(t, err) + assert.Equal(t, 0, stats.Added+stats.Updated, "unchanged docs must be skipped") + assert.Equal(t, 2, stats.Skipped) + + // Doc moves planned -> in_progress: status follows while task is planned. + writeDoc(t, repoRoot, ".doc/ops/long-term/"+slugPlanned+".md", + "---\ntitle: \"计划中任务 "+tag+"\"\nstatus: in_progress\n---\n\n# 计划中任务\n\n正文改。\n") + stats, err = im.Import(t.Context()) + require.NoError(t, err) + assert.Equal(t, 1, stats.Updated) + require.NoError(t, st.Pool.QueryRow(t.Context(), + `SELECT status, stage FROM cc_tasks WHERE slug = $1`, slugPlanned).Scan(&status, &stage)) + assert.Equal(t, pipeline.StatusInProgress, status) + assert.Equal(t, pipeline.StageProposal, stage, "stage must not move after insert") + + // Terminal status never regresses: doc edited to in_progress, task stays done. + writeDoc(t, repoRoot, ".doc/ops/short-term/"+slugDone+".md", + "---\ntitle: \"已完成任务 "+tag+"\"\nstatus: in_progress\nowner: cursor\nbranch: fix/x\nworktree: \"-\"\nmutex: docgov\n---\n\n# 已完成任务\n\n正文改。\n") + stats, err = im.Import(t.Context()) + require.NoError(t, err) + assert.Equal(t, 1, stats.Updated, "content changed but status must not regress") + require.NoError(t, st.Pool.QueryRow(t.Context(), + `SELECT status FROM cc_tasks WHERE slug = $1`, slugDone).Scan(&status)) + assert.Equal(t, pipeline.StatusDone, status) +} diff --git a/internal/controlcenter/server/export_doc_handler.go b/internal/controlcenter/server/export_doc_handler.go new file mode 100644 index 00000000..c090a8f9 --- /dev/null +++ b/internal/controlcenter/server/export_doc_handler.go @@ -0,0 +1,64 @@ +package server + +import ( + "errors" + "net/http" + "os" + "path/filepath" + "time" + + "chat-agent-system/internal/controlcenter/pipeline" + "chat-agent-system/internal/controlcenter/tasks" +) + +// handleTaskExportDoc renders a task as a .doc short-term task document and +// writes it to /.doc/ops/short-term/YYYY-MM-DD-.md. The +// periodic docs sync then indexes it like any other .doc file. An existing +// file is never overwritten (409). +func (s *Server) handleTaskExportDoc(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + t, err := tasks.GetTask(r.Context(), s.store, id) + if err != nil { + writeDomainError(w, err) + return + } + if !tasks.ValidDocSlug(t.Slug) { + writeError(w, http.StatusBadRequest, "task slug is not exportable: must match [a-z0-9-]+") + return + } + + today := time.Now().Format("2006-01-02") + rel := filepath.Join(".doc", "ops", "short-term", today+"-"+t.Slug+".md") + full := filepath.Join(s.cfg.RepoRoot, rel) + if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil { + writeDomainError(w, err) + return + } + f, err := os.OpenFile(full, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644) + if errors.Is(err, os.ErrExist) { + writeError(w, http.StatusConflict, "doc file already exists: "+filepath.ToSlash(rel)) + return + } + if err != nil { + writeDomainError(w, err) + return + } + if _, err := f.WriteString(tasks.RenderTaskDoc(t, today)); err != nil { + f.Close() + writeDomainError(w, err) + return + } + if err := f.Close(); err != nil { + writeDomainError(w, err) + return + } + + relSlash := filepath.ToSlash(rel) + t, err = tasks.PatchTask(r.Context(), s.store, id, tasks.Patch{DocPath: &relSlash}) + if err != nil { + writeDomainError(w, err) + return + } + s.appendEvent(r.Context(), id, "", pipeline.PhaseSystem, "exported doc "+relSlash, nil) + writeJSON(w, http.StatusCreated, map[string]any{"doc_path": relSlash, "task": t}) +} diff --git a/internal/controlcenter/server/export_doc_test.go b/internal/controlcenter/server/export_doc_test.go new file mode 100644 index 00000000..0619c136 --- /dev/null +++ b/internal/controlcenter/server/export_doc_test.go @@ -0,0 +1,85 @@ +package server + +import ( + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "chat-agent-system/internal/controlcenter/config" + "chat-agent-system/internal/controlcenter/store" +) + +func newExportTestServer(t *testing.T, repoRoot string) *httptest.Server { + t.Helper() + dsn := os.Getenv("TEST_CC_DB_DSN") + if dsn == "" { + t.Skip("TEST_CC_DB_DSN not set") + } + _, thisFile, _, _ := runtime.Caller(0) + migrationsDir := filepath.Join(filepath.Dir(thisFile), "..", "..", "..", "deploy", "control-center", "goose") + ctx := t.Context() + require.NoError(t, store.Migrate(ctx, dsn, migrationsDir)) + st, err := store.Connect(ctx, dsn) + require.NoError(t, err) + t.Cleanup(st.Close) + + srv := New(config.Config{ConsoleToken: testConsoleToken, RepoRoot: repoRoot}, st) + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(ts.Close) + return ts +} + +func TestTaskExportDoc(t *testing.T) { + repoRoot := t.TempDir() + ts := newExportTestServer(t, repoRoot) + + taskSlug := slug("export") + code, raw := doReq(t, "POST", ts.URL+"/api/v1/tasks", testConsoleToken, map[string]any{ + "slug": taskSlug, "title": "导出任务", "instruction_md": "## 目标\n\n验证导出。\n", + }) + require.Equal(t, http.StatusCreated, code, string(raw)) + id := decodeBody(t, raw)["id"].(string) + + code, raw = doReq(t, "POST", ts.URL+"/api/v1/tasks/"+id+"/export-doc", testConsoleToken, nil) + require.Equal(t, http.StatusCreated, code, string(raw)) + docPath := decodeBody(t, raw)["doc_path"].(string) + today := time.Now().Format("2006-01-02") + assert.Equal(t, ".doc/ops/short-term/"+today+"-"+taskSlug+".md", docPath) + + content, err := os.ReadFile(filepath.Join(repoRoot, filepath.FromSlash(docPath))) + require.NoError(t, err) + md := string(content) + assert.True(t, strings.HasPrefix(md, "---\n"), "front matter first") + assert.Contains(t, md, `title: "导出任务"`) + assert.Contains(t, md, "status: planned\n") + assert.Contains(t, md, "created: "+today+"\n") + assert.Contains(t, md, "# 导出任务") + assert.Contains(t, md, "## 目标") + + // task doc_path updated + code, raw = doReq(t, "GET", ts.URL+"/api/v1/tasks/"+id, testConsoleToken, nil) + require.Equal(t, http.StatusOK, code) + assert.Equal(t, docPath, decodeBody(t, raw)["doc_path"]) + + // second export conflicts, file untouched + code, raw = doReq(t, "POST", ts.URL+"/api/v1/tasks/"+id+"/export-doc", testConsoleToken, nil) + assert.Equal(t, http.StatusConflict, code, string(raw)) + assert.Contains(t, string(raw), "already exists") + + // unsafe slug rejected + code, raw = doReq(t, "POST", ts.URL+"/api/v1/tasks", testConsoleToken, map[string]any{ + "slug": slug("Bad_Slug"), "title": "不可导出", + }) + require.Equal(t, http.StatusCreated, code, string(raw)) + badID := decodeBody(t, raw)["id"].(string) + code, raw = doReq(t, "POST", ts.URL+"/api/v1/tasks/"+badID+"/export-doc", testConsoleToken, nil) + assert.Equal(t, http.StatusBadRequest, code, string(raw)) +} diff --git a/internal/controlcenter/server/server.go b/internal/controlcenter/server/server.go index 17d9ddda..4256b362 100644 --- a/internal/controlcenter/server/server.go +++ b/internal/controlcenter/server/server.go @@ -84,6 +84,7 @@ func (s *Server) routes() { s.mux.HandleFunc("POST /api/v1/tasks/{id}/stages/send-back", s.handleTaskSendBack) s.mux.HandleFunc("POST /api/v1/tasks/{id}/stages/{stage}/external-submit", s.handleTaskExternalSubmit) s.mux.HandleFunc("GET /api/v1/tasks/{id}/pipeline", s.handleTaskPipeline) + s.mux.HandleFunc("POST /api/v1/tasks/{id}/export-doc", s.handleTaskExportDoc) s.mux.HandleFunc("GET /api/v1/decisions", s.handleDecisionList) s.mux.HandleFunc("POST /api/v1/decisions/{id}/resolve", s.handleDecisionResolve) s.mux.HandleFunc("GET /api/v1/stage-settings", s.handleStageSettingsGet) diff --git a/internal/controlcenter/server/server_test.go b/internal/controlcenter/server/server_test.go index ea6415e7..c9a9ead4 100644 --- a/internal/controlcenter/server/server_test.go +++ b/internal/controlcenter/server/server_test.go @@ -108,7 +108,10 @@ func TestAgentHasNoAdvanceRoute(t *testing.T) { claimTask(t, ts, id) code, _ := doReq(t, "POST", ts.URL+"/api/v1/agent/tasks/"+id+"/stages/advance", testAgentToken, map[string]any{}) - assert.Equal(t, http.StatusNotFound, code, "agents must not have any stage-advance route") + // The SPA catch-all "GET /" makes the mux answer 405 (Allow: GET) for + // unmatched POST paths; either way the agent must have no advance route. + assert.Contains(t, []int{http.StatusNotFound, http.StatusMethodNotAllowed}, code, + "agents must not have any stage-advance route") // And the task is untouched. code, raw := doReq(t, "GET", ts.URL+"/api/v1/tasks/"+id, testConsoleToken, nil) diff --git a/internal/controlcenter/tasks/docgen.go b/internal/controlcenter/tasks/docgen.go new file mode 100644 index 00000000..741c1918 --- /dev/null +++ b/internal/controlcenter/tasks/docgen.go @@ -0,0 +1,58 @@ +package tasks + +import ( + "regexp" + "strconv" + "strings" + + "chat-agent-system/internal/controlcenter/pipeline" +) + +// docSlugRe constrains task slugs that may be exported to .doc files; it +// also makes path traversal impossible by construction. +var docSlugRe = regexp.MustCompile(`^[a-z0-9-]+$`) + +// ValidDocSlug reports whether slug is safe to use in a .doc file name. +func ValidDocSlug(slug string) bool { return docSlugRe.MatchString(slug) } + +// DocStatus maps a cc_tasks status back onto the .doc front-matter space. +func DocStatus(status string) string { + if status == pipeline.StatusBlockedDecision { + return "blocked" + } + return status +} + +// RenderTaskDoc renders a task as a .doc short-term task document following +// .doc/ops/templates/short-term-task.md. today is the YYYY-MM-DD date used +// for created/updated. The task's instruction_md becomes the body; when +// empty the template skeleton is used. +func RenderTaskDoc(t *Task, today string) string { + orDash := func(s string) string { + if strings.TrimSpace(s) == "" { + return "-" + } + return s + } + var b strings.Builder + b.WriteString("---\n") + b.WriteString("title: " + strconv.Quote(t.Title) + "\n") + b.WriteString("status: " + DocStatus(t.Status) + "\n") + b.WriteString("created: " + today + "\n") + b.WriteString("updated: " + today + "\n") + b.WriteString("owner: " + orDash(t.Owner) + "\n") + b.WriteString("branch: " + orDash(t.Branch) + "\n") + b.WriteString("worktree: " + strconv.Quote(orDash(t.Worktree)) + "\n") + b.WriteString("mutex: " + orDash(t.Mutex) + "\n") + b.WriteString("---\n\n") + + body := strings.TrimSpace(t.InstructionMD) + if body == "" { + body = "## 目标\n\n(一句话可验证目标)\n\n## 范围\n\n- \n\n## 非目标\n\n- \n\n" + + "## 步骤\n\n1. [ ] \n\n## 验收标准\n\n- [ ] \n\n## 相关链接\n\n" + + "- Checklist: \n- Handoff: \n- Test report: \n\n## 证据 / 备注\n\n(命令输出、commit、文件路径)" + } + b.WriteString("# " + t.Title + "\n\n") + b.WriteString(body + "\n") + return b.String() +} diff --git a/internal/controlcenter/tasks/tasks.go b/internal/controlcenter/tasks/tasks.go index a3181e0e..95beb16a 100644 --- a/internal/controlcenter/tasks/tasks.go +++ b/internal/controlcenter/tasks/tasks.go @@ -172,6 +172,7 @@ type Patch struct { Branch *string Worktree *string PlanID *string + DocPath *string } func PatchTask(ctx context.Context, st *store.Store, id string, p Patch) (*Task, error) { @@ -208,6 +209,9 @@ func PatchTask(ctx context.Context, st *store.Store, id string, p Patch) (*Task, if p.PlanID != nil { add("plan_id", *p.PlanID) } + if p.DocPath != nil { + add("doc_path", *p.DocPath) + } if len(sets) == 0 { return GetTask(ctx, st, id) }