Qualify snapshot-bound job transitions

apps/morphir-workbench/mck/README.md

@@ -2,7 +2,7 @@
 
 This component exposes the shipped `finos/morphir-execution` value codec and
 validator to the shared Rust MCK. It supports exact contracts
-`0.1.0-draft.1`, `0.1.0-draft.2`, `0.1.0-draft.3`, `0.1.0-draft.4` and `0.1.0-draft.5`. It has no goldens, evaluator or separate
+`0.1.0-draft.1` through `0.1.0-draft.6`. It has no goldens, evaluator or separate
 conformance runner. The parent `finos/morphir` repository owns the specification,
 corpora and verdict calculation. Draft.1/.2 are merged; draft.3 adds invocation
 admission; draft.4 adds output admission; draft.5 adds terminal outcome admission.
@@ -130,3 +130,34 @@ null/annotations, UTF-16 and value budgets. Earlier exact drafts remain supporte
 Admission of failed/cancelled evidence means valid evidence, not successful or
 authenticated execution. UI adoption, live daemon jobs, acknowledged cancellation
 and connected protocol extensions remain subsequent work.
+
+## Job transition qualification
+
+Exact offline MCK draft.6 advertises only `reduce-job` and JSON. It maps the
+unchanged `morphir/workbench/jobs` `0.1.0-draft.1` profile and its nested draft.5
+terminal outcomes. It does not add operations to the five historical drafts or
+define an Ion receipt mapping. Requalify historical kits separately.
+
+Input is exactly `{offer,identity,calls,initial,actions}`. The adapter admits the
+offer, caller-owned expected identity, ordered call IDs and initial submitted
+receipt through `finos/morphir-execution/jobs`. It preadmits closed action syntax
+before executing up to 64 actions. Actions are `observe` with a receipt,
+`reconnect` with an offer, `disconnect`, or `cancel-action`. Invalid receipt or
+offer payloads produce a bounded per-action rejection and leave the cursor
+unchanged. A malformed trace/action shape rejects the entire operation.
+
+The canonical observation retains the final identity, complete latest receipt,
+connection availability, cancellation action and ordered step observations.
+Each step includes its operation, disposition or rejection code, and a bounded
+view of sequence, state, acknowledgement, connection and cancellation action.
+Values and cleanup appear only in the final receipt, not in every view. Input
+and output are each capped at 1 MiB. The parent owns all fixed expectations,
+observation admission, comparisons and report rechecking. No expected data or
+case ID is sent to this adapter.
+
+Select `spec/workbench/mck/draft.6` with the existing gate. Set
+`MORPHIR_WORKBENCH_EVIDENCE` to retain reports in a separate directory per kit;
+the default remains `.dev/workbench-mck/`. This qualifies recorded transitions,
+replay/conflict handling, cancellation races and same-session reconciliation.
+It does not authenticate receipts or qualify live job RPC, host execution,
+durable recovery, Proton IPC or generated browser execution.

apps/morphir-workbench/mck/jobs.mbt

@@ -0,0 +1,167 @@
+///|
+priv enum JobAction {
+  Observe(Json)
+  Reconnect(Json)
+  Disconnect
+  CancelAction
+}
+
+///|
+fn JobAction::name(self : JobAction) -> String {
+  match self {
+    Observe(_) => "observe"
+    Reconnect(_) => "reconnect"
+    Disconnect => "disconnect"
+    CancelAction => "cancel-action"
+  }
+}
+
+///|
+fn job_action(value : Json) -> JobAction raise {
+  let names = match field(value, "op") {
+    String("observe") => ["op", "receipt"]
+    String("reconnect") => ["op", "offer"]
+    String("disconnect" | "cancel-action") => ["op"]
+    _ => raise @jobs.Invalid("workbench.invalid_job_trace")
+  }
+  ignore(closed_fields(value, names, "workbench.invalid_job_trace"))
+  match field(value, "op") {
+    String("observe") => Observe(field(value, "receipt"))
+    String("reconnect") => Reconnect(field(value, "offer"))
+    String("disconnect") => Disconnect
+    _ => CancelAction
+  }
+}
+
+///|
+fn cancel_action(cursor : @jobs.Cursor) -> String {
+  match cursor.cancel_action() {
+    Send => "send"
+    Pending => "pending"
+    AlreadyTerminal => "already-terminal"
+    Unavailable => "unavailable"
+    Disconnected => "disconnected"
+  }
+}
+
+///|
+fn job_view(cursor : @jobs.Cursor) -> Json raise {
+  let receipt = cursor.latest()
+  let state = match receipt.state() {
+    Submitted => "submitted"
+    Running => "running"
+    Terminal(_) => "terminal"
+  }
+  {
+    "sequence": receipt.sequence(),
+    "state": state,
+    "cancelRequested": receipt.cancel_requested(),
+    "connected": cursor.connected(),
+    "cancelAction": cancel_action(cursor),
+  }
+}
+
+///|
+fn job_step(
+  cursor : @jobs.Cursor,
+  action : JobAction,
+) -> (@jobs.Cursor, Json) raise {
+  let (next, status, code) : (@jobs.Cursor, String, String?) = try {
+    let (next, status) = match action {
+      Observe(value) => {
+        let (next, disposition) = cursor.observe(
+          @jobs.Receipt::from_json(value),
+        )
+        let status = match disposition {
+          Advanced => "advanced"
+          Replay => "replay"
+          Stale => "stale"
+        }
+        (next, status)
+      }
+      Reconnect(value) =>
+        (cursor.reconnect(@jobs.Offer::from_json(value)), "local")
+      Disconnect => (cursor.disconnect(), "local")
+      // Selecting a client action does not send a request or acknowledge it.
+      CancelAction => (cursor, "local")
+    }
+    (next, status, None)
+  } catch {
+    @jobs.Invalid(code) => (cursor, "invalid", Some(code))
+    @execution.Invalid(code) =>
+      (
+        cursor,
+        "invalid",
+        Some(
+          code.split(":").next().unwrap_or("workbench.codec_failure").to_owned(),
+        ),
+      )
+    // Nested Ion metadata decoders have their own error types. They are still
+    // rejected receipt payloads, and cannot erase earlier accepted evidence.
+    _ => (cursor, "invalid", Some("workbench.codec_failure"))
+  }
+  let observation : Json = {
+    "op": action.name(),
+    "status": status,
+    "view": job_view(next),
+  }
+  if code is Some(code) && observation is Object(fields) {
+    fields["code"] = Json::string(code)
+  }
+  (next, observation)
+}
+
+///|
+/// Offline traces call the shipped reducer. Expectations and verdicts stay in
+/// the parent MCK; no host authentication, execution or RPC is claimed here.
+fn reduce_job(input : Json, format : Json) -> Json raise {
+  if format != Json::string("json") {
+    raise @jobs.Invalid("workbench.invalid_request")
+  }
+  ignore(
+    closed_fields(
+      input,
+      ["offer", "identity", "calls", "initial", "actions"],
+      "workbench.invalid_job_trace",
+    ),
+  )
+  let offer = @jobs.Offer::from_json(field(input, "offer"))
+  let identity = @jobs.Identity::from_json(field(input, "identity"))
+  guard field(input, "calls") is Array(calls) else {
+    raise @jobs.Invalid("workbench.invalid_job_trace")
+  }
+  let calls = calls.map(value => {
+    guard value is String(name) else {
+      raise @jobs.Invalid("workbench.job_identity")
+    }
+    name
+  })
+  let initial = @jobs.Receipt::from_json(field(input, "initial"))
+  let mut cursor = @jobs.Cursor::start(offer, identity, calls, initial)
+  guard field(input, "actions") is Array(values) else {
+    raise @jobs.Invalid("workbench.invalid_job_trace")
+  }
+  if values.length() > 64 {
+    raise @jobs.Invalid("workbench.job_limit")
+  }
+  // Reject syntax for the whole trace before running any action. Receipt and
+  // offer payload admission happens per action and leaves the cursor intact.
+  let actions = values.map(job_action)
+  let observations = []
+  for action in actions {
+    let (next, observation) = job_step(cursor, action)
+    cursor = next
+    observations.push(observation)
+  }
+  let projection : Json = {
+    "identity": identity.to_json(),
+    "receipt": cursor.latest().to_json(),
+    "connected": cursor.connected(),
+    "cancelAction": cancel_action(cursor),
+    "observations": observations,
+  }
+  if @utf8.encode(projection.stringify()).length() > 1024 * 1024 {
+    raise @jobs.Invalid("workbench.job_limit")
+  }
+  projection
+}