From 7b29dbff7e43e87741476f47df17ad1582b16546 Mon Sep 17 00:00:00 2001 From: "Somhairle H. Marisol" Date: Mon, 21 Sep 2026 00:58:15 +0800 Subject: feat(pipeline): add runnable artifact pipeline increment [Change Nature] - This commit adds the first runnable increment of the F# artifact pipeline: CLI, server, frontend bundle, stage fixture, and tests. It is feature work, not a bug fix. [New Capability] - CLI `run` executes the three-step Blender fixture through the process bridge and writes GLBs, step logs, events, and a manifest; `verify` validates a manifest without Blender. - Server exposes run start, run-scoped SSE, artifact manifest/file routes, and serves the Fable frontend from its build output. - Frontend starts runs, follows run-scoped SSE progress, and links the artifact manifest on completion. [Implementation] - Shared layer holds domain, run state, replay, and artifact contracts; Modeling holds the pipeline, 120s bridge timeout, and result-file protocol; stage/blender/fixture.py owns geometry in Python (bpy owns geometry, F# owns orchestration). - Fable bundle is regenerated into public/ and copied by the server project; solution includes all projects with warnings-as-errors. - README documents build, Fable, server, interpreter/env, fixture limitations, and current verification status. [Impact] - Verified: 40/40 solution tests pass; Release build 0 warnings/errors; real CLI run with bpy 5.0.1 produced 3 GLBs, 3 logs, and a verify-clean manifest; isolated-port server smoke flow succeeded. - Not yet verified: browser end-to-end behavior and production fidelity. Artifacts and caches are gitignored. --- src/SomhairlesDream.Server/Store.fs | 259 ++++++++++++++++++++++++++++++++++++ 1 file changed, 259 insertions(+) create mode 100644 src/SomhairlesDream.Server/Store.fs (limited to 'src/SomhairlesDream.Server/Store.fs') diff --git a/src/SomhairlesDream.Server/Store.fs b/src/SomhairlesDream.Server/Store.fs new file mode 100644 index 0000000..b4a2293 --- /dev/null +++ b/src/SomhairlesDream.Server/Store.fs @@ -0,0 +1,259 @@ +namespace SomhairlesDream.Server + +open System +open System.Collections.Generic +open System.Threading.Channels +open SomhairlesDream.Shared + +type RunStore(projectId: string, runId: string, staleAfter: TimeSpan) = + let gate = obj () + let mutable state = RunState.initial projectId runId DateTimeOffset.UtcNow + let subscribers = ResizeArray>() + + let update transition = + lock gate (fun () -> + state <- transition state + for subscriber in subscribers do + subscriber.Writer.TryWrite(state) |> ignore + state) + + member _.Snapshot() = lock gate (fun () -> state) + + member _.Start(now: DateTimeOffset) = update (RunState.start now) + + member _.Heartbeat(now: DateTimeOffset) = update (RunState.heartbeat now) + + member _.Observe(now: DateTimeOffset) = update (RunState.statusAt staleAfter now) + + member _.Publish(now: DateTimeOffset, mesh: MeshSnapshot) = update (RunState.publish now mesh) + + member _.Subscribe() : ChannelReader = + let channel = Channel.CreateUnbounded() + + lock gate (fun () -> + channel.Writer.TryWrite(state) |> ignore + subscribers.Add(channel) + channel.Reader) + +type ArtifactRunSubscription(reader: ChannelReader, dispose: unit -> unit) = + let mutable disposed = false + + member _.Reader = reader + + member _.Dispose() = + if not disposed then + disposed <- true + dispose () + + interface IDisposable with + member this.Dispose() = this.Dispose() + +type ArtifactRunStore(ids: RunIds, staleAfter: TimeSpan, initialAt: DateTimeOffset) = + let gate = obj () + let subscribers = ResizeArray>() + let mutable state : RunSnapshot = + { ProjectId = ids.ProjectId + RunId = ids.RunId + BaseId = ids.BaseId + TargetId = ids.TargetId + Status = "idle" + CurrentStepId = None + CurrentStepIndex = -1 + TotalSteps = 0 + LastHeartbeatAt = None + UpdatedAt = initialAt + Message = "waiting for design run" + ManifestPath = None + ManifestUrl = None + Error = None + IsStale = false } + + let publish next = + for subscriber in subscribers do + subscriber.Writer.TryWrite(next) |> ignore + + let eventIds = function + | RunEvent.Start value -> value.Ids + | RunEvent.Heartbeat value -> value.Ids + | RunEvent.Checkpoint value -> value.Ids + | RunEvent.Complete value -> value.Ids + | RunEvent.Fail value -> value.Ids + + let eventName = function + | RunEvent.Start _ -> "start" + | RunEvent.Heartbeat _ -> "heartbeat" + | RunEvent.Checkpoint _ -> "checkpoint" + | RunEvent.Complete _ -> "complete" + | RunEvent.Fail _ -> "fail" + + let rejectIfNotRunning event = + if state.Status <> "running" then + Some $"cannot apply {eventName event} to run in status {state.Status}" + else + None + + let checkpointCount () = + if state.CurrentStepIndex < 0 then 0 else state.CurrentStepIndex + 1 + + let commit next = + state <- next + publish next + Ok next + + member _.Snapshot() = lock gate (fun () -> state) + + member _.Apply(event: RunEvent) : Result = + lock gate (fun () -> + if eventIds event <> ids then + Error "run event identity does not match the selected run" + else + match event with + | RunEvent.Start value -> + if state.Status <> "idle" then + Error $"cannot apply start to run in status {state.Status}" + elif value.TotalSteps <= 0 then + Error "run must contain at least one step" + else + commit + { state with + Status = "running" + TotalSteps = value.TotalSteps + CurrentStepId = None + CurrentStepIndex = -1 + LastHeartbeatAt = Some value.At + UpdatedAt = value.At + Message = "design run started" + Error = None + IsStale = false } + | RunEvent.Heartbeat value -> + match rejectIfNotRunning event with + | Some message -> Error message + | None -> + commit + { state with + LastHeartbeatAt = Some value.At + UpdatedAt = value.At + Message = value.Message + IsStale = false } + | RunEvent.Checkpoint value -> + match rejectIfNotRunning event with + | Some message -> Error message + | None when value.TotalSteps <> state.TotalSteps -> + Error "checkpoint total steps do not match run" + | None when value.StepIndex < 0 || value.StepIndex >= state.TotalSteps -> + Error "checkpoint step index is outside the run" + | None when value.StepIndex <> checkpointCount () -> + Error "checkpoint step index is out of order" + | None -> + commit + { state with + CurrentStepId = Some value.StepId + CurrentStepIndex = value.StepIndex + LastHeartbeatAt = Some value.At + UpdatedAt = value.At + Message = value.Message + IsStale = false } + | RunEvent.Complete value -> + match rejectIfNotRunning event with + | Some message -> Error message + | None when checkpointCount () <> state.TotalSteps -> + Error "cannot complete run before every step is checkpointed" + | None -> + commit + { state with + Status = "complete" + CurrentStepId = None + LastHeartbeatAt = Some value.At + UpdatedAt = value.At + Message = "design run complete" + ManifestPath = Some value.ManifestPath + ManifestUrl = Some $"/api/artifacts/manifest?projectId={ids.ProjectId}&runId={ids.RunId}" + Error = None + IsStale = false } + | RunEvent.Fail value -> + if state.Status <> "idle" && state.Status <> "running" then + Error $"cannot apply fail to run in status {state.Status}" + else + commit + { state with + Status = "failed" + CurrentStepId = value.StepId + LastHeartbeatAt = Some value.At + UpdatedAt = value.At + Message = value.Message + Error = Some value.Message + IsStale = false }) + + member _.Observe(now: DateTimeOffset) = + lock gate (fun () -> + let next = + match state.Status, state.LastHeartbeatAt with + | "running", Some heartbeat when now - heartbeat > staleAfter && not state.IsStale -> + Some + { state with + UpdatedAt = now + Message = "heartbeat timeout" + IsStale = true } + | "running", Some heartbeat when now - heartbeat <= staleAfter && state.IsStale -> + Some + { state with + UpdatedAt = now + Message = "heartbeat resumed" + IsStale = false } + | _ -> None + + match next with + | Some value -> + state <- value + publish value + value + | None -> state) + + member _.Subscribe() = + let channel = Channel.CreateUnbounded() + + lock gate (fun () -> + channel.Writer.TryWrite(state) |> ignore + subscribers.Add(channel)) + + new ArtifactRunSubscription( + channel.Reader, + fun () -> + lock gate (fun () -> + subscribers.Remove(channel) |> ignore + channel.Writer.TryComplete() |> ignore)) + +type ArtifactRunRegistry(staleAfter: TimeSpan, clock: unit -> DateTimeOffset) = + let gate = obj () + let runs = Dictionary() + + member _.TryCreate(ids: RunIds) = + lock gate (fun () -> + let key = ids.ProjectId, ids.RunId + + match runs.TryGetValue(key) with + | true, _ -> None + | false, _ -> + let store = ArtifactRunStore(ids, staleAfter, clock ()) + runs.Add(key, store) + Some store) + + member _.GetOrCreate(ids: RunIds) = + lock gate (fun () -> + let key = ids.ProjectId, ids.RunId + + match runs.TryGetValue(key) with + | true, store -> store + | false, _ -> + let store = ArtifactRunStore(ids, staleAfter, clock ()) + runs.Add(key, store) + store) + + member _.TryFind(projectId: string, runId: string) = + lock gate (fun () -> + match runs.TryGetValue((projectId, runId)) with + | true, store -> Some store + | false, _ -> None) + + member _.ObserveAll(now: DateTimeOffset) = + lock gate (fun () -> runs.Values |> Seq.map (fun store -> store.Observe(now)) |> Seq.toArray) -- cgit v1.2.3