orchestra/internal/federation/client.go

package federation

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"orchestra/internal/domain"
	"strings"
)

// Client is the worker-side protocol client. It carries no task state: the
// homesrv event log remains authoritative and workers only persist their
// local execution session/checkouts.
type Client struct {
	BaseURL    string
	WorkerID   string
	Token      string
	AdmitToken string
	HTTP       *http.Client
}

func (c Client) Register(ctx context.Context, w Worker) error {
	b, err := json.Marshal(map[string]any{"id": w.ID, "address": w.Address, "capacity": w.Capacity, "supported_projects": w.SupportedProjects, "build": w.Build, "token": c.Token})
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(c.BaseURL, "/")+"/v1/federation/workers", bytes.NewReader(b))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	if c.AdmitToken != "" {
		req.Header.Set("Authorization", "Bearer "+c.AdmitToken)
	}
	h := c.HTTP
	if h == nil {
		h = http.DefaultClient
	}
	resp, err := h.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		msg, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("federation register: %s: %s", resp.Status, strings.TrimSpace(string(msg)))
	}
	return nil
}

func (c Client) request(ctx context.Context, method, path string, body any) (*http.Response, error) {
	var r io.Reader
	if body != nil {
		b, err := json.Marshal(body)
		if err != nil {
			return nil, err
		}
		r = bytes.NewReader(b)
	}
	req, err := http.NewRequestWithContext(ctx, method, strings.TrimRight(c.BaseURL, "/")+path, r)
	if err != nil {
		return nil, err
	}
	req.Header.Set("X-Orchestra-Worker", c.WorkerID)
	req.Header.Set("Authorization", "Bearer "+c.Token)
	if body != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	h := c.HTTP
	if h == nil {
		h = http.DefaultClient
	}
	resp, err := h.Do(req)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode/100 != 2 {
		defer resp.Body.Close()
		b, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("federation: %s: %s", resp.Status, strings.TrimSpace(string(b)))
	}
	return resp, nil
}

func (c Client) Events(ctx context.Context, since uint64) ([]domain.Event, uint64, error) {
	resp, err := c.request(ctx, http.MethodGet, "/v1/federation/events?since="+fmt.Sprint(since), nil)
	if err != nil {
		return nil, 0, err
	}
	defer resp.Body.Close()
	var out struct {
		Cursor uint64         `json:"cursor"`
		Events []domain.Event `json:"events"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		return nil, 0, err
	}
	return out.Events, out.Cursor, nil
}

// Tasks hydrates the worker's cache when its local state predates the
// coordinator's event-retention window. The coordinator remains authoritative
// for the task projection.
func (c Client) Tasks(ctx context.Context) ([]domain.Task, error) {
	resp, err := c.request(ctx, http.MethodGet, "/v1/tasks", nil)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	var tasks []domain.Task
	if err := json.NewDecoder(resp.Body).Decode(&tasks); err != nil {
		return nil, err
	}
	return tasks, nil
}

// TurnDecision is the coordinator's answer at a worker's turn boundary: the
// verdict the worker reported, plus the human decisions this session has not
// been shown. Decisions are present only when the verdict is continue.
type TurnDecision struct {
	Verdict   string                 `json:"verdict"`
	Decisions []domain.HumanDecision `json:"decisions,omitempty"`
}

// Turn reports a verified turn boundary and collects any newer human
// decisions. The worker evaluates rotation locally, because only it can see
// the pane; authority stays with the coordinator.
func (c Client) Turn(ctx context.Context, taskID, epoch, verdict string, delivered []string) (TurnDecision, error) {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/turn", map[string]any{
		"task_id": taskID, "lease_epoch": epoch, "verdict": verdict, "delivered_decisions": delivered,
	})
	if err != nil {
		return TurnDecision{}, err
	}
	defer resp.Body.Close()
	var out TurnDecision
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		return TurnDecision{}, err
	}
	return out, nil
}

// AdvancePhase carries an agent's bounded phase-change request to the
// coordinator, which decides. The accepted phase comes back so the worker
// knows the session it owns has been superseded and must rotate.
//
// artifact is the sealed output of the phase being left, and is empty for a
// phase that produces none.
func (c Client) AdvancePhase(ctx context.Context, taskID, epoch, operationID string, from, to domain.WorkPhase, artifact []byte) (domain.WorkPhase, error) {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/phase", map[string]any{
		"task_id": taskID, "lease_epoch": epoch, "operation_id": operationID,
		"from": string(from), "to": string(to), "artifact": artifact,
	})
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	var out struct {
		Phase domain.WorkPhase `json:"phase"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		return "", err
	}
	return out.Phase, nil
}

// Intent fetches the reduced authority for one task: its contract plus the
// human decisions still standing. A worker renders its launch instruction
// from this, never from handoff prose.
func (c Client) Intent(ctx context.Context, taskID string) (domain.EffectiveIntent, error) {
	resp, err := c.request(ctx, http.MethodGet, "/v1/tasks/"+url.PathEscape(taskID)+"/intent", nil)
	if err != nil {
		return domain.EffectiveIntent{}, err
	}
	defer resp.Body.Close()
	var intent domain.EffectiveIntent
	if err := json.NewDecoder(resp.Body).Decode(&intent); err != nil {
		return domain.EffectiveIntent{}, err
	}
	return intent, nil
}

func (c Client) Ack(ctx context.Context, cursor uint64) error {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/events/ack", map[string]uint64{"cursor": cursor})
	if resp != nil {
		resp.Body.Close()
	}
	return err
}
func (c Client) Heartbeat(ctx context.Context, health WorkerHealth) error {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/heartbeat", health)
	if resp != nil {
		resp.Body.Close()
	}
	return err
}
func (c Client) Renew(ctx context.Context, taskID, epoch string, expectedVersion, ttlSeconds int) error {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/renew", map[string]any{"task_id": taskID, "lease_epoch": epoch, "expected_version": expectedVersion, "ttl_seconds": ttlSeconds})
	if resp != nil {
		resp.Body.Close()
	}
	return err
}
func (c Client) Start(ctx context.Context, taskID, epoch string, expectedVersion int, evidence domain.SessionEvidence) error {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/start", map[string]any{"task_id": taskID, "lease_epoch": epoch, "expected_version": expectedVersion, "session_evidence": evidence})
	if resp != nil {
		resp.Body.Close()
	}
	return err
}
func (c Client) NackStart(ctx context.Context, taskID, epoch string, expectedVersion int, failureClass, detail string, evidence domain.SessionEvidence) error {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/nack", map[string]any{"task_id": taskID, "lease_epoch": epoch, "expected_version": expectedVersion, "failure_class": failureClass, "last_error": detail, "session_evidence": evidence})
	if resp != nil {
		resp.Body.Close()
	}
	return err
}
func (c Client) Artifact(ctx context.Context, ref string) ([]byte, error) {
	resp, err := c.request(ctx, http.MethodGet, "/v1/artifacts/"+url.PathEscape(ref), nil)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	return io.ReadAll(resp.Body)
}

func (c Client) PutArtifact(ctx context.Context, b []byte) (string, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(c.BaseURL, "/")+"/v1/artifacts", bytes.NewReader(b))
	if err != nil {
		return "", err
	}
	req.Header.Set("Content-Type", "application/octet-stream")
	req.Header.Set("X-Orchestra-Worker", c.WorkerID)
	req.Header.Set("Authorization", "Bearer "+c.Token)
	h := c.HTTP
	if h == nil {
		h = http.DefaultClient
	}
	resp, err := h.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		msg, _ := io.ReadAll(resp.Body)
		return "", fmt.Errorf("artifact upload: %s: %s", resp.Status, strings.TrimSpace(string(msg)))
	}
	var out struct {
		Ref string `json:"ref"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		return "", err
	}
	return out.Ref, nil
}
func (c Client) Release(ctx context.Context, taskID, ref, anchor, transactionID, epoch string, expectedVersion int, evidence domain.SessionEvidence) error {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/handoff", map[string]any{"task_id": taskID, "handoff_ref": ref, "anchor_sha": anchor, "transaction_id": transactionID, "lease_epoch": epoch, "expected_version": expectedVersion, "session_evidence": evidence})
	if resp != nil {
		resp.Body.Close()
	}
	return err
}
func (c Client) Pickup(ctx context.Context, taskID, ref, anchor, transactionID, epoch string, leaseVersion int, evidence domain.SessionEvidence) error {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/pickup", map[string]any{"task_id": taskID, "handoff_ref": ref, "anchor_sha": anchor, "transaction_id": transactionID, "lease_epoch": epoch, "lease_version": leaseVersion, "session_evidence": evidence})
	if resp != nil {
		resp.Body.Close()
	}
	return err
}
func (c Client) Complete(ctx context.Context, taskID, reportRef, resultSHA, branch, remote, epoch string, expectedVersion int, receipt map[string]any, evidence domain.SessionEvidence) error {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/complete", map[string]any{"task_id": taskID, "handoff_ref": reportRef, "result_sha": resultSHA, "branch": branch, "remote": remote, "lease_epoch": epoch, "expected_version": expectedVersion, "receipt": receipt, "session_evidence": evidence})
	if resp != nil {
		resp.Body.Close()
	}
	return err
}

func (c Client) PublishCapture(ctx context.Context, capture Capture) (Capture, error) {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/captures", capture)
	if err != nil {
		return Capture{}, err
	}
	defer resp.Body.Close()
	var out Capture
	err = json.NewDecoder(resp.Body).Decode(&out)
	return out, err
}
func (c Client) Commands(ctx context.Context) ([]Command, error) {
	resp, err := c.request(ctx, http.MethodGet, "/v1/federation/commands", nil)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	var out []Command
	err = json.NewDecoder(resp.Body).Decode(&out)
	return out, err
}
func (c Client) ResolveCommand(ctx context.Context, id, status, message string) error {
	resp, err := c.request(ctx, http.MethodPost, "/v1/federation/commands/"+url.PathEscape(id), map[string]string{"status": status, "message": message})
	if resp != nil {
		resp.Body.Close()
	}
	return err
}