diff --git a/apps/morphir-workbench/mck/README.md b/apps/morphir-workbench/mck/README.md index 7f221e54..af95b321 100644 --- a/apps/morphir-workbench/mck/README.md +++ b/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. diff --git a/apps/morphir-workbench/mck/jobs.mbt b/apps/morphir-workbench/mck/jobs.mbt new file mode 100644 index 00000000..f07178ad --- /dev/null +++ b/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 +} diff --git a/apps/morphir-workbench/mck/jobs_test.mbt b/apps/morphir-workbench/mck/jobs_test.mbt new file mode 100644 index 00000000..b02e997d --- /dev/null +++ b/apps/morphir-workbench/mck/jobs_test.mbt @@ -0,0 +1,303 @@ +///| +fn job_identity() -> Json { + { + "hostId": "host", + "sessionId": "session", + "providerId": "provider", + "jobId": "job", + "snapshotId": "snapshot", + } +} + +///| +fn job_offer() -> Json { + { + "profile": "morphir/workbench/jobs", + "contractVersion": "0.1.0-draft.1", + "hostId": "host", + "sessionId": "session", + "providerId": "provider", + "operations": ["query", "cancel"], + } +} + +///| +fn job_receipt( + sequence : Int, + state : String, + acknowledged? : Bool = false, + outcomes? : Json = Json::null(), +) -> Json { + { + "contractVersion": "0.1.0-draft.1", + "identity": job_identity(), + "sequence": sequence, + "state": state, + "cancelRequested": acknowledged, + "outcomes": outcomes, + } +} + +///| +fn job_request(actions : Array[Json]) -> Json { + { + "id": 1, + "op": "reduce-job", + "suite": "workbench", + "contractVersion": "0.1.0-draft.6", + "format": "json", + "input": { + "offer": job_offer(), + "identity": job_identity(), + "calls": ["call"], + "initial": job_receipt(0, "submitted"), + "actions": actions, + }, + } +} + +///| +fn job_field(value : Json, name : String) -> Json { + match value { + Object(fields) => fields[name] + _ => abort("expected object") + } +} + +///| +fn job_items(value : Json) -> Array[Json] { + match value { + Array(items) => items + _ => abort("expected array") + } +} + +///| +test "job draft negotiates only JSON reduction and preserves historical claims" { + assert_eq( + @mck.handle({ + "id": 1, + "op": "capabilities", + "suite": "workbench", + "contractVersion": "0.1.0-draft.6", + }), + { + "id": 1, + "suite": "workbench", + "contractVersion": "0.1.0-draft.6", + "binding": "morphir-moonbit", + "language": "moonbit", + "operations": ["reduce-job"], + "formats": ["json"], + }, + ) + for + version in [ + "0.1.0-draft.1", "0.1.0-draft.2", "0.1.0-draft.3", "0.1.0-draft.4", "0.1.0-draft.5", + ] { + let request : Json = { + "id": 1, + "op": "reduce-job", + "suite": "workbench", + "contractVersion": version, + "format": "json", + "input": {}, + } + assert_eq(@mck.handle(request), { + "id": 1, + "status": "invalid", + "code": "workbench.invalid_request", + }) + } +} + +///| +test "job transcript retains state after rejected evidence and can continue" { + let result = @mck.handle( + job_request([ + { "op": "observe", "receipt": job_receipt(2, "running") }, + { "op": "observe", "receipt": job_receipt(3, "submitted") }, + { "op": "observe", "receipt": job_receipt(1, "submitted") }, + { "op": "observe", "receipt": job_receipt(2, "running") }, + { + "op": "observe", + "receipt": job_receipt(5, "running", acknowledged=true), + }, + { "op": "cancel-action" }, + ]), + ) + assert_eq(job_field(result, "status"), Json::string("ok")) + let value = job_field(result, "value") + let observations = job_items(job_field(value, "observations")) + assert_eq(job_field(observations[0], "status"), Json::string("advanced")) + assert_eq( + job_field(observations[1], "code"), + Json::string("workbench.job_transition"), + ) + assert_eq( + job_field(observations[1], "view"), + job_field(observations[0], "view"), + ) + assert_eq(job_field(observations[2], "status"), Json::string("stale")) + assert_eq(job_field(observations[3], "status"), Json::string("replay")) + assert_eq( + job_field(value, "receipt"), + job_receipt(5, "running", acknowledged=true), + ) + assert_eq(job_field(value, "cancelAction"), Json::string("pending")) +} + +///| +test "offline observation cannot manufacture a host terminal" { + let result = @mck.handle( + job_request([ + { "op": "disconnect" }, + { "op": "observe", "receipt": job_receipt(1, "running") }, + { "op": "cancel-action" }, + { "op": "reconnect", "offer": job_offer() }, + { "op": "observe", "receipt": job_receipt(1, "running") }, + ]), + ) + let value = job_field(result, "value") + let steps = job_items(job_field(value, "observations")) + assert_eq( + job_field(steps[1], "code"), + Json::string("workbench.job_disconnected"), + ) + assert_eq(job_field(steps[2], "view"), job_field(steps[0], "view")) + assert_eq(job_field(value, "receipt"), job_receipt(1, "running")) + assert_eq(job_field(value, "connected"), Json::boolean(true)) +} + +///| +test "returned data and cleanup survive an acknowledged cancellation race" { + let outcomes : Json = { + "contractVersion": "0.1.0-draft.5", + "calls": [ + { + "id": "call", + "outcome": { + "status": "returned", + "value": { "type": "int", "value": "0009007199254740993" }, + }, + }, + ], + "cleanup": { + "stage": "cleanup", + "code": "provider.dispose", + "message": "cleanup", + }, + "extensions": null, + } + let result = @mck.handle( + job_request([ + { + "op": "observe", + "receipt": job_receipt(1, "running", acknowledged=true), + }, + { + "op": "observe", + "receipt": job_receipt(2, "terminal", acknowledged=true, outcomes~), + }, + { "op": "disconnect" }, + ]), + ) + let value = job_field(result, "value") + assert_eq(job_field(value, "cancelAction"), Json::string("already-terminal")) + assert_eq(job_field(value, "connected"), Json::boolean(false)) + let terminal = job_field(job_field(value, "receipt"), "outcomes") + assert_eq(job_field(terminal, "cleanup"), job_field(outcomes, "cleanup")) + let call = job_items(job_field(terminal, "calls"))[0] + assert_eq( + job_field(job_field(job_field(call, "outcome"), "value"), "value"), + Json::string("9007199254740993"), + ) +} + +///| +test "job action syntax is closed and bounded before any step runs" { + assert_eq( + @mck.handle(job_request([{ "op": "cancel-action", "extra": true }])), + { "id": 1, "status": "invalid", "code": "workbench.invalid_job_trace" }, + ) + assert_eq( + @mck.handle(job_request(Array::make(65, { "op": "cancel-action" }))), + { "id": 1, "status": "invalid", "code": "workbench.job_limit" }, + ) + assert_eq(@mck.handle(job_request([{ "op": "observe", "receipt": null }])), { + "id": 1, + "status": "ok", + "value": { + "identity": job_identity(), + "receipt": job_receipt(0, "submitted"), + "connected": true, + "cancelAction": "send", + "observations": [ + { + "op": "observe", + "status": "invalid", + "code": "workbench.job_receipt", + "view": { + "sequence": 0, + "state": "submitted", + "cancelRequested": false, + "connected": true, + "cancelAction": "send", + }, + }, + ], + }, + }) +} + +///| +test "job raw JSON duplicate members cannot disappear at the boundary" { + let line = "{\"id\":1,\"op\":\"reduce-job\",\"suite\":\"workbench\",\"contractVersion\":\"0.1.0-draft.6\",\"format\":\"json\",\"input\":{\"offer\":{},\"offer\":{}}}" + assert_eq( + @mck.handle_line(line), + Some({ "id": 1, "status": "invalid", "code": "workbench.invalid_request" }), + ) +} + +///| +test "job input budget applies before malformed trace admission" { + let request : Json = { + "id": 1, + "op": "reduce-job", + "suite": "workbench", + "contractVersion": "0.1.0-draft.6", + "format": "json", + "input": { "padding": "x".repeat(1024 * 1024) }, + } + assert_eq(@mck.handle(request), { + "id": 1, + "status": "invalid", + "code": "workbench.input_limit", + }) +} + +///| +test "invalid nested Ion metadata rejects only its action and retains cursor" { + let outcomes : Json = { + "contractVersion": "0.1.0-draft.5", + "calls": [{ "id": "call", "outcome": { "status": "cancelled" } }], + "cleanup": null, + "extensions": { "format": "ion-binary", "bytes": [255] }, + } + let result = @mck.handle( + job_request([ + { "op": "observe", "receipt": job_receipt(1, "terminal", outcomes~) }, + { "op": "cancel-action" }, + ]), + ) + assert_eq(job_field(result, "status"), Json::string("ok")) + let value = job_field(result, "value") + let steps = job_items(job_field(value, "observations")) + assert_eq( + job_field(steps[0], "code"), + Json::string("workbench.codec_failure"), + ) + assert_eq(job_field(steps[1], "status"), Json::string("local")) + assert_eq(job_field(steps[1], "view"), job_field(steps[0], "view")) + assert_eq(job_field(value, "receipt"), job_receipt(0, "submitted")) +} diff --git a/apps/morphir-workbench/mck/moon.pkg b/apps/morphir-workbench/mck/moon.pkg index 11d7ab29..f0cd8f08 100644 --- a/apps/morphir-workbench/mck/moon.pkg +++ b/apps/morphir-workbench/mck/moon.pkg @@ -1,5 +1,6 @@ import { "finos/morphir-execution" @execution, + "finos/morphir-execution/jobs", "moonrockz/ion/text", "moonrockz/ion/ion", "moonbitlang/core/json", diff --git a/apps/morphir-workbench/mck/protocol.mbt b/apps/morphir-workbench/mck/protocol.mbt index 8334e544..66379dc8 100644 --- a/apps/morphir-workbench/mck/protocol.mbt +++ b/apps/morphir-workbench/mck/protocol.mbt @@ -75,6 +75,7 @@ pub fn handle(request : Json) -> Json { ["id", "op", "suite", "contractVersion", "format", "input"] } let version = field(request, "contractVersion") + let draft6 = version == Json::string("0.1.0-draft.6") let draft2 = version == Json::string("0.1.0-draft.2") let draft3 = version == Json::string("0.1.0-draft.3") let draft5 = version == Json::string("0.1.0-draft.5") @@ -82,7 +83,13 @@ pub fn handle(request : Json) -> Json { guard fields.length() == allowed.length() && fields.keys().all(name => allowed.contains(name)) && field(request, "suite") == Json::string("workbench") && - (version == Json::string("0.1.0-draft.1") || draft2 || draft3 || draft4) else { + ( + version == Json::string("0.1.0-draft.1") || + draft2 || + draft3 || + draft4 || + draft6 + ) else { return invalid(id, "workbench.invalid_request") } if op == Json::string("capabilities") { @@ -92,7 +99,9 @@ pub fn handle(request : Json) -> Json { "contractVersion": version, "binding": "morphir-moonbit", "language": "moonbit", - "operations": if draft5 { + "operations": if draft6 { + ["reduce-job"] + } else if draft5 { [ "decode-value", "validate-value", "validate-invocations", "validate-output", "validate-outcomes", @@ -108,23 +117,35 @@ pub fn handle(request : Json) -> Json { } else { ["decode-value"] }, - "formats": ["json", "ion-text"], + "formats": if draft6 { + ["json"] + } else { + ["json", "ion-text"] + }, } } - guard op == Json::string("decode-value") || - ((draft2 || draft3 || draft4) && op == Json::string("validate-value")) || - ((draft3 || draft4) && op == Json::string("validate-invocations")) || - (draft4 && op == Json::string("validate-output")) || - (draft5 && op == Json::string("validate-outcomes")) else { + guard (draft6 && op == Json::string("reduce-job")) || + ( + !draft6 && + ( + op == Json::string("decode-value") || + ((draft2 || draft3 || draft4) && op == Json::string("validate-value")) || + ((draft3 || draft4) && op == Json::string("validate-invocations")) || + (draft4 && op == Json::string("validate-output")) || + (draft5 && op == Json::string("validate-outcomes")) + ) + ) else { return invalid(id, "workbench.invalid_request") } try { let input = field(request, "input") - if (draft2 || draft3 || draft4) && + if (draft2 || draft3 || draft4 || draft6) && @utf8.encode(input.stringify()).length() > 1024 * 1024 { return invalid(id, "workbench.input_limit") } - let observation = if op == Json::string("validate-outcomes") { + let observation = if draft6 { + reduce_job(input, field(request, "format")) + } else if op == Json::string("validate-outcomes") { admit_outcomes(input, field(request, "format")).to_json() } else if op == Json::string("validate-invocations") { admit_invocations(input, field(request, "format")).to_json() @@ -143,6 +164,7 @@ pub fn handle(request : Json) -> Json { } { "id": id, "status": "ok", "value": observation } } catch { + @jobs.Invalid(code) => invalid(id, code) @execution.Invalid(code) => { // Legacy codec diagnostics append input paths/names after a colon. // The MCK observation carries only the bounded, stable rejection code. diff --git a/apps/morphir-workbench/mck/protocol_test.mbt b/apps/morphir-workbench/mck/protocol_test.mbt index 0b7686d9..e7050d3b 100644 --- a/apps/morphir-workbench/mck/protocol_test.mbt +++ b/apps/morphir-workbench/mck/protocol_test.mbt @@ -91,7 +91,7 @@ test "adapter rejects unsupported drafts and leaked expected values" { "id": 3, "op": "capabilities", "suite": "workbench", - "contractVersion": "0.1.0-draft.6", + "contractVersion": "0.1.0-draft.7", }, { "id": 3, diff --git a/scripts/workbench-mck.mbtx b/scripts/workbench-mck.mbtx index 5b4bbb30..f0027d9d 100644 --- a/scripts/workbench-mck.mbtx +++ b/scripts/workbench-mck.mbtx @@ -1,5 +1,5 @@ ///| -/// Opt-in value, invocation, output and terminal qualification against a selected parent-owned MCK corpus +/// Opt-in value, invocation, output, terminal and job-trace qualification against a selected parent-owned MCK corpus /// and Rust runner. No upstream checkout or toolchain is acquired by this gate. import { "moonbitlang/async@0.22.4", @@ -47,6 +47,10 @@ async fn main { @stdio.stderr.write("The selected corpus has no cases.\n") @sys.exit(1) } + let evidence_root = @env.get_env_var("MORPHIR_WORKBENCH_EVIDENCE").unwrap_or( + ".dev/workbench-mck", + ) + @fs.mkdir(evidence_root, recursive=true) for target in ["js", "native", "wasm", "wasm-gc"] { checked("moon", ["test", "apps/morphir-workbench/mck", "--target", target]) } @@ -60,7 +64,7 @@ async fn main { ) let artifact = "_build/\{target}/\{mode}/build/finos/morphir-workbench/mck/\{target}/\{target}" + (if target == "js" { ".js" } else { ".exe" }) - let report = ".dev/workbench-mck/\{target}-\{mode}.json" + let report = "\{evidence_root}/\{target}-\{mode}.json" let adapter_args = if target == "js" { ["--adapter", "node", "--adapter-arg", artifact] } else {