multica/server/internal/qoderruntime/state.go

package qoderruntime

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"time"
)

// One journal belongs to exactly one bridge configuration and holds every
// active run, bounded by the connection's concurrency limit. Credentials are
// never persisted. Keep this directory on persistent storage.
type journal struct {
	Binding   string   `json:"binding"`
	RuntimeID string   `json:"runtime_id"`
	Cleanup   []string `json:"cleanup,omitempty"`
	Runs      []*run   `json:"runs,omitempty"`
	// Run is the single-run format written before concurrency; decode folds it into Runs.
	Run *run `json:"run,omitempty"`
}
type run struct {
	// PriorSessionID is the session of this agent's previous run on the same
	// issue or chat.
	// An idle prior session is continued instead of creating a new one.
	PriorSessionID     string           `json:"prior_session_id,omitempty"`
	Continued          bool             `json:"continued,omitempty"`
	Deadline           time.Time        `json:"deadline"`
	SubmissionDeadline time.Time        `json:"submission_deadline,omitempty"`
	RetryAt            time.Time        `json:"retry_at,omitempty"`
	AgentID            string           `json:"agent_id,omitempty"`
	Page               string           `json:"page,omitempty"`
	TaskID             string           `json:"task_id"`
	SessionID          string           `json:"session_id,omitempty"`
	Prompt             string           `json:"prompt,omitempty"`
	Resources          []map[string]any `json:"resources,omitempty"`
	Phase              string           `json:"phase"`
	Cursor             string           `json:"cursor,omitempty"`
	Seq                int              `json:"seq"`
	Started            bool             `json:"started"`
	Output             string           `json:"output,omitempty"`
	Failure            string           `json:"failure,omitempty"`
	Outcome            string           `json:"outcome,omitempty"`
}

func save(path string, j *journal) error {
	b, err := json.Marshal(j)
	if err != nil {
		return err
	}
	f, err := os.CreateTemp(filepath.Dir(path), ".qoder-state-*")
	if err != nil {
		return err
	}
	defer os.Remove(f.Name())
	if _, err = f.Write(b); err != nil {
		f.Close()
		return err
	}
	if err = f.Sync(); err != nil {
		f.Close()
		return err
	}
	if err = f.Close(); err != nil {
		return err
	}
	if err = os.Rename(f.Name(), path); err != nil {
		return err
	}
	return syncStateDir(filepath.Dir(path))
}
func readJournal(path, binding string) (*journal, error) {
	b, err := os.ReadFile(path)
	if os.IsNotExist(err) {
		return &journal{Binding: binding}, nil
	}
	if err != nil {
		return nil, err
	}
	return decodeJournal(b, binding)
}
func decodeJournal(b []byte, binding string) (*journal, error) {
	if len(b) == 0 {
		return &journal{Binding: binding}, nil
	}
	var j journal
	if err := json.Unmarshal(b, &j); err != nil {
		return nil, fmt.Errorf("invalid bridge journal: %w", err)
	}
	if j.Binding != binding {
		return nil, fmt.Errorf("state directory belongs to a different bridge configuration")
	}
	if j.Binding == "" {
		j.Binding = binding
	}
	if j.Run != nil {
		j.Runs = append(j.Runs, j.Run)
		j.Run = nil
	}
	seen := map[string]bool{}
	for _, r := range j.Runs {
		if r == nil || r.TaskID == "" {
			return nil, fmt.Errorf("bridge journal has no task ID")
		}
		if seen[r.TaskID] {
			return nil, fmt.Errorf("bridge journal has a duplicate task")
		}
		seen[r.TaskID] = true
		switch r.Phase {
		case "prepared", "creating", "final":
		case "created", "sending", "running":
			if r.SessionID == "" {
				return nil, fmt.Errorf("bridge journal has no session ID")
			}
		default:
			return nil, fmt.Errorf("unknown bridge journal phase")
		}
	}
	return &j, nil
}

func (j *journal) remove(r *run) {
	for i, existing := range j.Runs {
		if existing == r {
			j.Runs = append(j.Runs[:i], j.Runs[i+1:]...)
			return
		}
	}
}

type persistenceError struct{ err error }

func (e *persistenceError) Error() string { return "persist bridge journal: " + e.err.Error() }
func (b *Bridge) persist() error {
	if b.cfg.SaveState != nil {
		raw, err := json.Marshal(b.state)
		if err != nil {
			return &persistenceError{err}
		}
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if err := b.cfg.SaveState(ctx, raw); err != nil {
			return &persistenceError{err}
		}
		return nil
	}
	if err := save(b.path, b.state); err != nil {
		return &persistenceError{err}
	}
	return nil
}