Agent-System/internal/controlcenter/importer/importer.go

// 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
}