majordomo/provider/llamaswap/videochain.go

// videochain.go implements videogen.ChainerProvider against the videoutils
// chain orchestrator reached through llama-swap's /upstream passthrough
// (ADR-0025):
//
//	POST /upstream/<id>/v1/video/chain          JSON submit -> {job_id}
//	GET  /upstream/<id>/v1/jobs/{id}            -> {status,segment,total,segments}
//	GET  /upstream/<id>/v1/jobs/{id}/result     -> encoded clip
//	GET  /upstream/<id>/v1/jobs/{id}/segments/{n} -> encoded clip
//
// Unlike the ACE-Step music path this client does NOT hide the job queue
// behind a blocking call: a chain runs through multiple GPU swaps for many
// minutes, and the caller (mort's long-video tool) owns the poll loop so it
// can deliver PARTIAL results — completed segments survive a mid-chain
// failure and stay fetchable via ChainSegmentResult.
package llamaswap

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"strings"

	"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
	"gitea.stevedudenhoeffer.com/steve/majordomo/videogen"
)

// ChainerModel implements videogen.ChainerProvider. The id selects which
// upstream llama-swap loads (the videoutils orchestrator, which in turn
// drives the generation model through llama-swap itself).
func (p *Provider) ChainerModel(id string, opts ...videogen.ChainerModelOption) (videogen.Chainer, error) {
	if err := p.requireBaseURL(); err != nil {
		return nil, err
	}
	_ = videogen.ApplyChainerModelOptions(opts)
	return &chainerModel{p: p, id: id}, nil
}

type chainerModel struct {
	p  *Provider
	id string
}

// chainSubmitRequest is the videoutils POST /v1/video/chain shape. The init
// image rides as base64 in the JSON body (`init_image_b64`) — the submit is
// JSON, not multipart, per the pinned host contract.
type chainSubmitRequest struct {
	Segments     []chainSegmentWire `json:"segments"`
	InitImageB64 string             `json:"init_image_b64,omitempty"`
	SmoothJoins  bool               `json:"smooth_joins,omitempty"`
	Size         string             `json:"size,omitempty"`
}

type chainSegmentWire struct {
	Prompt  string  `json:"prompt"`
	Seconds float64 `json:"seconds,omitempty"`
}

// SubmitChain implements videogen.Chainer.
func (m *chainerModel) SubmitChain(ctx context.Context, req videogen.ChainRequest) (string, error) {
	if len(req.Segments) == 0 {
		return "", fmt.Errorf("%w: video chain requires at least one segment", llm.ErrUnsupported)
	}
	wire := chainSubmitRequest{
		SmoothJoins: req.SmoothJoins,
		Size:        strings.TrimSpace(req.Size),
	}
	for i, seg := range req.Segments {
		if strings.TrimSpace(seg.Prompt) == "" {
			return "", fmt.Errorf("%w: video chain segment %d requires a prompt", llm.ErrUnsupported, i)
		}
		if seg.Seconds < 0 {
			return "", fmt.Errorf("%w: video chain segment %d seconds must be >= 0, got %g", llm.ErrUnsupported, i, seg.Seconds)
		}
		wire.Segments = append(wire.Segments, chainSegmentWire{Prompt: seg.Prompt, Seconds: seg.Seconds})
	}
	if len(req.InitImage) > 0 {
		wire.InitImageB64 = base64.StdEncoding.EncodeToString(req.InitImage)
	}
	path, err := upstreamPath(m.id, "/v1/video/chain")
	if err != nil {
		return "", err
	}
	// Tolerant envelope: {"job_id": ...} per the contract, with data-wrapped
	// and bare-id fallbacks (musicgen release_task precedent).
	var resp struct {
		JobID string `json:"job_id"`
		ID    string `json:"id"`
		Data  struct {
			JobID string `json:"job_id"`
			ID    string `json:"id"`
		} `json:"data"`
	}
	if err := m.p.doJSON(ctx, http.MethodPost, path, m.id, &wire, &resp); err != nil {
		return "", err
	}
	for _, id := range []string{resp.JobID, resp.Data.JobID, resp.ID, resp.Data.ID} {
		if id != "" {
			return id, nil
		}
	}
	return "", &llm.APIError{Provider: m.p.name, Model: m.id, Message: "video chain submit returned no job_id"}
}

// chainJobResponse is the GET /v1/jobs/{id} shape. `segments` entries are
// tolerated as bare strings or objects keyed by id/segment_id.
type chainJobResponse struct {
	Status   string            `json:"status"`
	Segment  int               `json:"segment"`
	Total    int               `json:"total"`
	Segments []json.RawMessage `json:"segments"`
}

// ChainStatus implements videogen.Chainer.
func (m *chainerModel) ChainStatus(ctx context.Context, jobID string) (*videogen.ChainJob, error) {
	path, err := m.jobPath(jobID, "")
	if err != nil {
		return nil, err
	}
	var raw json.RawMessage
	if err := m.p.doJSON(ctx, http.MethodGet, path, m.id, nil, &raw); err != nil {
		return nil, err
	}
	var out chainJobResponse
	if err := json.Unmarshal(raw, &out); err != nil {
		return nil, fmt.Errorf("llama-swap: decode chain job status: %w", err)
	}
	job := &videogen.ChainJob{
		Status:  out.Status,
		Segment: out.Segment,
		Total:   out.Total,
		Raw:     raw,
	}
	for _, entry := range out.Segments {
		var s string
		if json.Unmarshal(entry, &s) == nil {
			job.SegmentIDs = append(job.SegmentIDs, s)
			continue
		}
		var obj struct {
			ID        string `json:"id"`
			SegmentID string `json:"segment_id"`
		}
		if json.Unmarshal(entry, &obj) == nil {
			switch {
			case obj.ID != "":
				job.SegmentIDs = append(job.SegmentIDs, obj.ID)
			case obj.SegmentID != "":
				job.SegmentIDs = append(job.SegmentIDs, obj.SegmentID)
			}
		}
	}
	if job.Status == "" {
		return nil, &llm.APIError{Provider: m.p.name, Model: m.id,
			Message: "chain job status payload carried no status: " + truncateForError(raw)}
	}
	return job, nil
}

// ChainResult implements videogen.Chainer.
func (m *chainerModel) ChainResult(ctx context.Context, jobID string) (*videogen.Result, error) {
	path, err := m.jobPath(jobID, "/result")
	if err != nil {
		return nil, err
	}
	raw, respType, err := m.p.doRaw(ctx, http.MethodGet, path, m.id, "", nil, maxVideoResponseBytes)
	if err != nil {
		return nil, err
	}
	return singleVideoResult(m.p.name, m.id, "video chain result", raw, respType)
}

// ChainSegmentResult implements videogen.Chainer.
func (m *chainerModel) ChainSegmentResult(ctx context.Context, jobID string, n int) (*videogen.Result, error) {
	if n < 0 {
		return nil, fmt.Errorf("%w: chain segment index must be >= 0, got %d", llm.ErrUnsupported, n)
	}
	path, err := m.jobPath(jobID, "/segments/"+strconv.Itoa(n))
	if err != nil {
		return nil, err
	}
	raw, respType, err := m.p.doRaw(ctx, http.MethodGet, path, m.id, "", nil, maxVideoResponseBytes)
	if err != nil {
		return nil, err
	}
	return singleVideoResult(m.p.name, m.id, "video chain segment", raw, respType)
}

// jobPath builds /upstream/<model>/v1/jobs/<jobID><suffix>, refusing job ids
// that carry path structure. The id is SERVER-SUPPLIED (echoed back from
// SubmitChain), so like upstreamPath's rest-component checks this rejects
// rather than escapes — a hostile/buggy upstream must not be able to steer
// the follow-up request at another proxy endpoint.
func (m *chainerModel) jobPath(jobID, suffix string) (string, error) {
	if strings.TrimSpace(jobID) == "" {
		return "", fmt.Errorf("llama-swap: chain job call requires a job id")
	}
	if strings.ContainsAny(jobID, "/?#") || strings.Contains(jobID, "..") {
		return "", fmt.Errorf("llama-swap: invalid chain job id %q (contains path structure)", jobID)
	}
	return upstreamPath(m.id, "/v1/jobs/"+jobID+suffix)
}