Documentation
¶
Overview ¶
Package uploadflow is the upload session state machine, extracted from the `melange model upload` command so that non-CLI frontends (the MCP server's upload_model tool) can drive the same flow.
The package is deliberately presentation-free: it never prints, never imports cobra, and reports outcomes through the Events interface, typed errors, and the Result. Frontends (the cobra adapter in internal/cmd/model, the MCP tool) own all final message formatting, state file cleanup on terminal outcomes, and closing the session Lease.
The durable resume-state and GCS primitives live in internal/upload and are reused as-is.
Index ¶
- Constants
- func BuildSpecs(in ManifestInputs, events Events) ([]upload.FileSpec, []upload.BucketSpec, error)
- func CompletedModelJSON(body []byte) (json.RawMessage, error)
- func ManifestOptions(specs []upload.BucketSpec) *gen.ManifestOptions
- func RecoverableCompletionState(state string) bool
- func TerminalCompletionWithoutModel(out *gen.CompleteModelUploadResponse) bool
- type ConflictError
- type Events
- type ManifestInputs
- type NopEvents
- type Orchestrator
- type Phase
- type Request
- type Result
- type ResumeOptions
- type SessionError
- type TerminalStateError
- type UsageError
Constants ¶
const ( SessionStateCreated = "CREATED" SessionStateUploading = "UPLOADING" SessionStateVerifying = "VERIFYING" SessionStateDispatchPending = "DISPATCH_PENDING" )
Upload-session states that retain the repository's single active slot (ADR-5 vocabulary, compared case-insensitively).
Variables ¶
This section is empty.
Functions ¶
func BuildSpecs ¶
func BuildSpecs(in ManifestInputs, events Events) ([]upload.FileSpec, []upload.BucketSpec, error)
BuildSpecs digests the local files named by in. Usage-shaped problems (missing model file, duplicate basenames, invalid manifests) surface as *UsageError. The returned buckets echo in.Buckets or, for a manifest document, its declared buckets.
func CompletedModelJSON ¶
func CompletedModelJSON(body []byte) (json.RawMessage, error)
CompletedModelJSON extracts the raw model object from an upload-complete response body, byte-exact.
func ManifestOptions ¶
func ManifestOptions(specs []upload.BucketSpec) *gen.ManifestOptions
ManifestOptions converts validated local bucket declarations to the exact OpenAPI wire shape. A nil pointer omits options for ordinary models.
func RecoverableCompletionState ¶
RecoverableCompletionState reports whether a session state means all bytes are server-owned and completion can simply be replayed.
func TerminalCompletionWithoutModel ¶
func TerminalCompletionWithoutModel(out *gen.CompleteModelUploadResponse) bool
TerminalCompletionWithoutModel reports a completion response that is terminal yet carries no model reference: such a session can never yield a model and a new upload must be started.
Types ¶
type ConflictError ¶
ConflictError reports a create rejected because an upload session already holds the repository's single active slot. SessionID/State identify the holder when resolution succeeded (SessionID may be empty). Stale means the conflicting session turned terminal during resolution and a retry already happened; Err is the original API conflict error.
func (*ConflictError) Error ¶
func (e *ConflictError) Error() string
func (*ConflictError) Unwrap ¶
func (e *ConflictError) Unwrap() error
type Events ¶
type Events interface {
// Progress reports one file's transfer progress. committed grows with
// server-acknowledged bytes; committed == total is emitted exactly once
// per file and signals that file's completion.
Progress(file string, committed, total int64)
// Note reports one fully formatted, human-readable flow message
// (without a trailing newline).
Note(msg string)
}
Events observes flow progress. Implementations render (CLI stderr) or log (MCP); the flow never writes to any stream itself.
type ManifestInputs ¶
type ManifestInputs struct {
ModelFile string
Inputs []string
External []string
InputManifest string
// Buckets are the already-parsed .pt2 shape buckets (the CLI's --bucket
// flags); only valid together with ModelFile.
Buckets []upload.BucketSpec
}
ManifestInputs names the local files of one upload, mirroring the CLI's upload flag vocabulary. ModelFile is ignored when InputManifest is set; exactly one of the two must be provided.
type Orchestrator ¶
type Orchestrator struct {
// Gen is the generated API client over the authenticated transport.
Gen *gen.ClientWithResponses
// Events receives progress and notes; nil discards them.
Events Events
// Bare is the HTTP client used against signed GCS URLs. It must carry
// NO API transport chain (no Authorization header, no debug logging):
// signed URLs and resumable session URIs are credentials.
Bare *http.Client
// StallTimeout is the per-chunk inactivity budget during transfers.
StallTimeout time.Duration
// Poll seams for completion polling: nil selects the real
// jitter/sleeper/clock in internal/wait.
Jitter func(time.Duration) time.Duration
Sleep func(context.Context, time.Duration) error
Now func() time.Time
// TransferSleep, when non-nil, replaces the GCS uploader's retry
// backoff sleep (tests inject it; the CLI leaves it nil).
TransferSleep func(context.Context, time.Duration) error
}
Orchestrator drives upload sessions: create → transfer (resume/reissue) → complete. Zero-value seams select production behavior.
func (*Orchestrator) Resume ¶
func (o *Orchestrator) Resume(ctx context.Context, sessionID string, opts ResumeOptions) (*Result, error)
Resume continues the session: replay completion when the server already owns every byte, otherwise reconcile local state (rebuilding it from the server when missing or corrupt) and transfer the remainder. See Result for the partial-result-with-Lease contract on errors.
type Phase ¶
type Phase int
Phase names where in the flow an error occurred, so frontends can attach phase-appropriate remediation.
const ( // PhaseTransfer covers byte transfer to signed URLs (the session is // preserved and resumable; acknowledged bytes are never re-sent). PhaseTransfer Phase = iota + 1 // PhaseComplete covers session completion (the session is preserved; // completion can be replayed via resume). PhaseComplete )
type Request ¶
type Request struct {
Account string
Name string
// Repo is ACCOUNT/NAME as displayed and persisted in resume state.
Repo string
Specs []upload.FileSpec
Buckets []upload.BucketSpec
// Wait polls completion (with deliberate replays) until a model
// reference or terminal state is observable, within Timeout.
Wait bool
Timeout time.Duration
}
Request describes one fresh upload: the target repository and the already-digested local manifest (see BuildSpecs).
type Result ¶
type Result struct {
SessionID string
Repo string
// Response is the typed final completion response (nil on partial
// results).
Response *gen.CompleteModelUploadResponse
// Completion is Response's raw body, byte-exact.
Completion json.RawMessage
// Model is the raw "model" object from Completion; nil when absent.
Model json.RawMessage
// WaitStarted is the completion clock start, for callers sharing the
// Wait budget with follow-up polling.
WaitStarted time.Time
Lease *upload.SessionLease
}
Result is the outcome of a completed flow. Completion carries the raw upload-complete response body byte-exact; Model carries the raw model object from it (nil when the response has none).
Lease is the held cross-process session lock: the caller owns closing it on EVERY non-nil Result. On errors after the lock was acquired, Run and Resume return a partial Result (SessionID, Repo, Lease) alongside the error so the session identifiers and the lock reach the caller.
func (*Result) CloseLease ¶
CloseLease releases the session lock the Result carries. It is nil-safe on both the Result and the Lease so frontends can defer it unconditionally right where Run or Resume returns — the invariant is "non-nil Result ⇔ lock held", and this helper makes honoring it on every path a one-liner.
type ResumeOptions ¶
type ResumeOptions struct {
Account string
Name string
Repo string
// BuildSpecs lazily digests the local files for a state rebuild; it is
// invoked only when the local state file is missing or corrupt. nil
// means the caller provided no local file arguments.
BuildSpecs func() ([]upload.FileSpec, error)
Wait bool
Timeout time.Duration
}
ResumeOptions carries what Resume needs beyond the session id.
type SessionError ¶
SessionError wraps a transfer or completion failure of a preserved, resumable session. Err is the underlying cause (context.Canceled for interrupts, wait.ErrTimeout for an exhausted Wait budget).
func (*SessionError) Error ¶
func (e *SessionError) Error() string
func (*SessionError) Unwrap ¶
func (e *SessionError) Unwrap() error
type TerminalStateError ¶
TerminalStateError reports a resume against a session whose server-side state is terminal: it can never be resumed. The caller should discard any local resume state for the session.
func (*TerminalStateError) Error ¶
func (e *TerminalStateError) Error() string
type UsageError ¶
type UsageError struct{ Err error }
UsageError marks input-shaped problems (the CLI maps them to usage errors, exit 2).
func (*UsageError) Error ¶
func (e *UsageError) Error() string
func (*UsageError) Unwrap ¶
func (e *UsageError) Unwrap() error