quarks-workflow-engine

The quarks-workflow-engine worked example.

Run it from sema/:

sema check examples/quarks-workflow-engine
SEMA_STRICT=1 sema run examples/quarks-workflow-engine
sema assure examples/quarks-workflow-engine --grade silver

Source

src/main.sema

"""Quarks 15-phase RunService vertical slice on native Sema agents and a durable circuit.

Ports the pinned Quarks engine semantics (upstream 6ec748e25c00d89a2a66ff4d6228a0cb333c32c1)
over the frozen research-default 15-phase graph: graph-order frontier scheduling,
review-driven macro iteration (invalidate back to the earliest named loop target and
re-drive with a compacted iteration context), per-passed-phase workspace checkpoints,
restart + rehydrate without re-executing committed phases, and the upstream best-effort
settle when no loop budget or usable redrive target remains.

Profiles:
  default  matches pinned upstream: an exhausted loop budget or unusable redrive
           target settles BEST-EFFORT — the review is committed and the run proceeds
           to packaging and COMPLETES.
  strict   named Sema safety profile: the same two conditions fail closed with a
           typed terminal run state instead of completing best-effort.

Each scenario emits one canonical JSON line (sorted keys, compact separators) that is
byte-comparable with oracle/quarks_upstream_oracle.py, which EXECUTES the pinned
upstream RunService. The deterministic mock model seam is the in-module
@provides("agent.execute") executor; no live provider is required.
"""

assure silver

# -- frozen graph: a data projection of app/assets/graphs/research-default/graph.yaml --

def phase_order() -> list[str] !{}:
    return [
        "user_input",
        "knowledge_acquisition",
        "knowledge_distillation",
        "literature_review",
        "hypothesis_methodology",
        "user_presentation",
        "derive_math_methodology",
        "experiment_design",
        "validation_simulation",
        "visualization_synthesis",
        "insights_refinement",
        "writing_presentation",
        "revision",
        "review_feedback",
        "packaging_release",
    ]

def deps_of(phase: str) -> list[str] !{}:
    require phase in phase_order()
    if phase == "user_input":
        return []
    if phase == "knowledge_acquisition":
        return ["user_input"]
    if phase == "knowledge_distillation":
        return ["user_input", "knowledge_acquisition"]
    if phase == "literature_review":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation"]
    if phase == "hypothesis_methodology":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review"]
    if phase == "user_presentation":
        return ["user_input", "knowledge_distillation", "hypothesis_methodology"]
    if phase == "derive_math_methodology":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review", "hypothesis_methodology", "user_presentation"]
    if phase == "experiment_design":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review", "hypothesis_methodology", "user_presentation", "derive_math_methodology"]
    if phase == "validation_simulation":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review", "hypothesis_methodology", "user_presentation", "experiment_design", "derive_math_methodology"]
    if phase == "visualization_synthesis":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review", "hypothesis_methodology", "user_presentation", "derive_math_methodology", "experiment_design", "validation_simulation"]
    if phase == "insights_refinement":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review", "hypothesis_methodology", "user_presentation", "experiment_design", "derive_math_methodology", "validation_simulation", "visualization_synthesis"]
    if phase == "writing_presentation":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review", "hypothesis_methodology", "user_presentation", "experiment_design", "validation_simulation", "visualization_synthesis", "insights_refinement", "derive_math_methodology"]
    if phase == "revision":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review", "hypothesis_methodology", "experiment_design", "validation_simulation", "visualization_synthesis", "writing_presentation", "insights_refinement", "derive_math_methodology"]
    if phase == "review_feedback":
        return ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review", "hypothesis_methodology", "insights_refinement", "writing_presentation", "derive_math_methodology", "experiment_design", "validation_simulation", "visualization_synthesis", "revision"]
    return ["user_input", "knowledge_acquisition", "validation_simulation", "visualization_synthesis", "writing_presentation", "review_feedback"]

# -- shared fixture constants (mirrored verbatim by the upstream oracle) --

def fix_plan_fixture() -> list[str] !{}:
    return [
        "Add an ablation over the core hyperparameters.",
        "Add the missing baseline comparison to the experiments.",
    ]

def loop_reason_fixture() -> str !{}:
    return "The manuscript needs another revision pass."

# -- deterministic mock model seam --

def phase_result(scenario: str, phase: str, iteration: int) -> str !{}:
    """One phase turn: 'verdict|decision|targets' — the decision table is the
    deterministic mock of the review model; every phase passes its contract."""
    if phase != "review_feedback":
        return "passed|none|"
    if scenario == "redrive_after_invalidation" and iteration == 0:
        return "passed|iterate|hypothesis_methodology,experiment_design"
    if scenario == "best_effort_budget_exhausted" or scenario == "strict_budget_exhausted":
        return "passed|iterate|hypothesis_methodology,experiment_design"
    if scenario == "best_effort_no_target" or scenario == "strict_no_target":
        return "passed|iterate|unknown_alpha,unknown_beta"
    return "passed|approve|"

@provides("agent.execute")
def scripted_harness(packet: dict[str, any]) -> str !{}:
    return phase_result(
        packet["inputs"]["scenario"],
        packet["inputs"]["phase"],
        packet["inputs"]["iteration"],
    )

agent execute_phase(scenario: str, phase: str, iteration: int) -> str by workflow_model:
    sem "Execute one Quarks research phase deterministically and report its verdict line"
    budget model_calls=1, tokens=128
    ensure len(result) > 0

# -- pure engine core: scheduler, invalidation, list helpers --

def appended(items: list[str], item: str) -> list[str] !{}:
    mut result = []
    for value in items:
        result.append(value)
    result.append(item)
    return result

def appended_unique(items: list[str], item: str) -> list[str] !{}:
    if item in items:
        return items
    return appended(items, item)

def copied(items: list[str]) -> list[str] !{}:
    mut result = []
    for value in items:
        result.append(value)
    return result

def first_n(items: list[str], count: int) -> list[str] !{}:
    require count >= 0
    mut prefix = []
    for value in items:
        if len(prefix) < count:
            prefix.append(value)
    return prefix

def subset_of(items: list[str], container: list[str]) -> bool !{}:
    for item in items:
        if item in container:
            continue
        return false
    return true

def is_complete(completed: list[str]) -> bool !{}:
    return subset_of(phase_order(), completed)

def next_ready(completed: list[str]) -> str !{}:
    for phase in phase_order():
        if phase in completed:
            continue
        if subset_of(deps_of(phase), completed):
            return phase
    return ""

def earliest_target(targets: list[str]) -> str !{}:
    for phase in phase_order():
        if phase in targets:
            return phase
    return ""

def dropped_after_invalidation(frontier: str, completed: list[str]) -> list[str] !{}:
    """The frontier plus its transitive downstream among the completed phases, in
    completed order (a candidate drops when the frontier or an already-dropped
    phase is among its dependencies)."""
    require frontier in phase_order()
    mut dropped = []
    for phase in completed:
        if phase == frontier:
            dropped.append(phase)
            continue
        mut hit = false
        for dep in deps_of(phase):
            if dep == frontier or dep in dropped:
                hit = true
        if hit:
            dropped.append(phase)
    return dropped

def kept_after_invalidation(dropped: list[str], completed: list[str]) -> list[str] !{}:
    mut kept = []
    for phase in completed:
        if phase in dropped:
            continue
        kept.append(phase)
    return kept

# -- canonical JSON rendering (sorted keys, compact separators) --

def jstr(text: str) -> str !{}:
    return "\"" + text + "\""

def jlist(items: list[str]) -> str !{}:
    return "[" + ",".join(items) + "]"

def jstrs(items: list[str]) -> str !{}:
    mut rendered = []
    for item in items:
        rendered.append(jstr(item))
    return jlist(rendered)

def render_execution(iteration: int, phase: str, segment: int, seq: int) -> str !{}:
    mut out = "{\"iteration\":" + str(iteration)
    out = out + ",\"phase\":" + jstr(phase)
    out = out + ",\"segment\":" + str(segment)
    out = out + ",\"seq\":" + str(seq)
    return out + ",\"verdict\":\"passed\"}"

def render_invalidation(dropped: list[str], frontier: str, iteration: int, kept: list[str]) -> str !{}:
    mut out = "{\"dropped\":" + jstrs(dropped)
    out = out + ",\"frontier\":" + jstr(frontier)
    out = out + ",\"iteration\":" + str(iteration)
    return out + ",\"kept\":" + jstrs(kept) + "}"

def render_context(iteration: int, prior: list[str], resume_target: str, targets: list[str]) -> str !{}:
    mut out = "{\"fix_plan\":" + jstrs(fix_plan_fixture())
    out = out + ",\"iteration\":" + str(iteration)
    out = out + ",\"loop_reason\":" + jstr(loop_reason_fixture())
    out = out + ",\"prior_output_phases\":" + jstrs(prior)
    out = out + ",\"resume_target_phase\":" + jstr(resume_target)
    return out + ",\"targets\":" + jstrs(targets) + "}"

def render_rehydration(checkpoint: str, segment: int) -> str !{}:
    return "{\"checkpoint\":" + jstr(checkpoint) + ",\"segment\":" + str(segment) + "}"

def render_settle(kind: str, reason: str) -> str !{}:
    if reason == "":
        return "{\"kind\":" + jstr(kind) + "}"
    return "{\"kind\":" + jstr(kind) + ",\"reason\":" + jstr(reason) + "}"

def render_trace(scenario: str, checkpoints: list[str], completed: list[str], executions: list[str], invalidations: list[str], contexts: list[str], loop_budget: int, loop_iterations: int, rehydrations: list[str], settle: str, status: str) -> str !{}:
    mut out = "{\"checkpoints\":" + jstrs(checkpoints)
    out = out + ",\"completed_phases\":" + jstrs(completed)
    out = out + ",\"executions\":" + jlist(executions)
    out = out + ",\"invalidations\":" + jlist(invalidations)
    out = out + ",\"iteration_contexts\":" + jlist(contexts)
    out = out + ",\"loop_budget\":" + str(loop_budget)
    out = out + ",\"loop_iterations\":" + str(loop_iterations)
    out = out + ",\"rehydrations\":" + jlist(rehydrations)
    out = out + ",\"scenario\":" + jstr(scenario)
    out = out + ",\"settle\":" + settle
    return out + ",\"status\":" + jstr(status) + "}"

# -- the durable run driver --

def run_scenario(scenario: str, profile: str, loop_budget: int, seg1_budget: int) -> str !{model.invoke}:
    """Drive one run of the 15-phase graph under the pinned engine semantics.

    ``seg1_budget`` > 0 splits the drive into two segments with a process-restart
    boundary between them: live workspace materials die at the boundary and MUST
    come back from the recorded checkpoint (rehydration) for the second segment's
    dependency gate to pass. Committed phases are never re-executed.
    """
    require loop_budget >= 0 and loop_budget <= 8
    require seg1_budget >= 0 and seg1_budget <= 15
    require profile == "default" or profile == "strict"
    rehydration = seg1_budget > 0
    mut completed = []
    mut outputs = []
    mut workspace = []
    mut executions = []
    mut invalidations = []
    mut contexts = []
    mut checkpoints = []
    mut checkpoint_ws = []
    mut checkpoint_label = ""
    mut rehydrations = []
    mut loop_iterations = 0
    mut settle = render_settle("none", "")
    mut status = "running"
    mut segment = 1
    mut seq = 0
    mut steps = 0
    while status == "running" and steps < 80:
        steps = steps + 1
        if segment == 1 and seg1_budget > 0 and seq >= seg1_budget:
            # Process restart analog: the live workspace dies with the segment; the
            # recorded checkpoint is the ONLY way the next segment sees the prior
            # phases' materials. Fail closed when no checkpoint was recorded.
            segment = 2
            workspace = []
            ensure checkpoint_label != ""
            workspace = copied(checkpoint_ws)
            rehydrations.append(render_rehydration(checkpoint_label, segment))
        phase = next_ready(completed)
        if phase == "":
            ensure is_complete(completed)
            status = "completed"
            continue
        # Fail-closed materials gate: every dependency's explored materials must be
        # live in the workspace (fresh execution or rehydrated checkpoint).
        ensure subset_of(deps_of(phase), workspace)
        seq = seq + 1
        turn = execute_phase(scenario, phase, loop_iterations)
        parts = turn.split("|")
        ensure len(parts) == 3
        ensure parts[0] == "passed"
        executions.append(render_execution(loop_iterations, phase, segment, seq))
        outputs = appended_unique(outputs, phase)
        workspace = appended_unique(workspace, phase)
        mut redriven = false
        if phase == "review_feedback" and parts[1] == "iterate":
            targets = parts[2].split(",")
            if loop_budget <= 0 or loop_iterations >= loop_budget:
                if profile == "strict":
                    status = "failed"
                    settle = render_settle("fail_closed", "budget_exhausted")
                    continue
                settle = render_settle("best_effort", "budget_exhausted")
            else:
                frontier = earliest_target(targets)
                if frontier == "":
                    if profile == "strict":
                        status = "failed"
                        settle = render_settle("fail_closed", "no_usable_target")
                        continue
                    settle = render_settle("best_effort", "no_usable_target")
                else:
                    iteration_index = loop_iterations + 1
                    dropped = dropped_after_invalidation(frontier, completed)
                    kept = kept_after_invalidation(dropped, completed)
                    invalidations.append(render_invalidation(dropped, frontier, iteration_index, kept))
                    contexts.append(render_context(iteration_index, sorted(outputs), frontier, targets))
                    completed = kept
                    loop_iterations = iteration_index
                    redriven = true
        if redriven:
            continue
        completed = appended(completed, phase)
        if rehydration:
            label = str(len(checkpoints) + 1) + "@" + phase
            checkpoints.append(label)
            checkpoint_ws = copied(workspace)
            checkpoint_label = label
        if phase == "packaging_release":
            status = "completed"
    ensure status != "running"
    return render_trace(scenario, checkpoints, completed, executions, invalidations, contexts, loop_budget, loop_iterations, rehydrations, settle, status)

circuit run_suite() -> list[str] !{model.invoke}:
    budget agents=192, spawn_depth=0, model_calls=192, tokens=65536
    mut lines = []
    lines.append(run_scenario("happy_path", "default", 0, 0))
    lines.append(run_scenario("redrive_after_invalidation", "default", 3, 0))
    lines.append(run_scenario("snapshot_restart_rehydrate", "default", 0, 7))
    lines.append(run_scenario("best_effort_budget_exhausted", "default", 1, 0))
    lines.append(run_scenario("best_effort_no_target", "default", 3, 0))
    lines.append(run_scenario("strict_budget_exhausted", "strict", 1, 0))
    lines.append(run_scenario("strict_no_target", "strict", 3, 0))
    return lines

def main() -> str !{model.invoke}:
    return "\n".join(run_suite())

# -- tests: the pure engine core --

test "scheduler walks the frozen graph order on a linear drive":
    ensure next_ready([]) == "user_input"
    mut done = []
    for phase in phase_order():
        ensure next_ready(done) == phase
        done = appended(done, phase)
    ensure next_ready(done) == ""
    ensure is_complete(done)

test "invalidation drops the frontier plus its transitive downstream in completed order":
    completed = first_n(phase_order(), 13)
    dropped = dropped_after_invalidation("hypothesis_methodology", completed)
    ensure dropped == [
        "hypothesis_methodology",
        "user_presentation",
        "derive_math_methodology",
        "experiment_design",
        "validation_simulation",
        "visualization_synthesis",
        "insights_refinement",
        "writing_presentation",
        "revision",
    ]
    kept = kept_after_invalidation(dropped, completed)
    ensure kept == ["user_input", "knowledge_acquisition", "knowledge_distillation", "literature_review"]
    ensure next_ready(kept) == "hypothesis_methodology"

test "earliest loop target follows graph order and unknown targets are unusable":
    ensure earliest_target(["experiment_design", "hypothesis_methodology"]) == "hypothesis_methodology"
    ensure earliest_target(["unknown_alpha", "unknown_beta"]) == ""

test "review decision table is deterministic per scenario and iteration":
    ensure phase_result("redrive_after_invalidation", "review_feedback", 0) == "passed|iterate|hypothesis_methodology,experiment_design"
    ensure phase_result("redrive_after_invalidation", "review_feedback", 1) == "passed|approve|"
    ensure phase_result("best_effort_budget_exhausted", "review_feedback", 5) == "passed|iterate|hypothesis_methodology,experiment_design"
    ensure phase_result("happy_path", "user_input", 0) == "passed|none|"

test "canonical renderers emit sorted-key compact json":
    ensure render_execution(0, "user_input", 1, 1) == "{\"iteration\":0,\"phase\":\"user_input\",\"segment\":1,\"seq\":1,\"verdict\":\"passed\"}"
    ensure render_settle("none", "") == "{\"kind\":\"none\"}"
    ensure render_settle("best_effort", "budget_exhausted") == "{\"kind\":\"best_effort\",\"reason\":\"budget_exhausted\"}"
    ensure render_rehydration("7@derive_math_methodology", 2) == "{\"checkpoint\":\"7@derive_math_methodology\",\"segment\":2}"

Reflected API

main

Quarks 15-phase RunService vertical slice on native Sema agents and a durable circuit.

Ports the pinned Quarks engine semantics (upstream 6ec748e25c00d89a2a66ff4d6228a0cb333c32c1) over the frozen research-default 15-phase graph: graph-order frontier scheduling, review-driven macro iteration (invalidate back to the earliest named loop target and re-drive with a compacted iteration context), per-passed-phase workspace checkpoints, restart + rehydrate without re-executing committed phases, and the upstream best-effort settle when no loop budget or usable redrive target remains.

Profiles: default matches pinned upstream: an exhausted loop budget or unusable redrive target settles BEST-EFFORT — the review is committed and the run proceeds to packaging and COMPLETES. strict named Sema safety profile: the same two conditions fail closed with a typed terminal run state instead of completing best-effort.

Each scenario emits one canonical JSON line (sorted keys, compact separators) that is byte-comparable with oracle/quarks_upstream_oracle.py, which EXECUTES the pinned upstream RunService. The deterministic mock model seam is the in-module @provides(“agent.execute”) executor; no live provider is required.

def phase_order

def phase_order() -> list[str] !{}

Returns list[str]

Effects !{}

def deps_of

def deps_of(phase: str) -> list[str] !{}

Parameters

name	type
phase	str

Returns list[str]

Effects !{}

def fix_plan_fixture

def fix_plan_fixture() -> list[str] !{}

Returns list[str]

Effects !{}

def loop_reason_fixture

def loop_reason_fixture() -> str !{}

Returns str

Effects !{}

def phase_result

def phase_result(scenario: str, phase: str, iteration: int) -> str !{}

Parameters

name	type
scenario	str
phase	str
iteration	int

Returns str

Effects !{}

One phase turn: ‘verdict|decision|targets’ — the decision table is the deterministic mock of the review model; every phase passes its contract.

def scripted_harness

def scripted_harness(packet: dict[str, any]) -> str !{}

Parameters

name	type
packet	dict[str, any]

Returns str

Effects !{}

agent execute_phase

agent execute_phase(scenario: str, phase: str, iteration: int) -> str

Parameters

name	type
scenario	str
phase	str
iteration	int

Returns str

def appended

def appended(items: list[str], item: str) -> list[str] !{}

Parameters

name	type
items	list[str]
item	str

Returns list[str]

Effects !{}

def appended_unique

def appended_unique(items: list[str], item: str) -> list[str] !{}

Parameters

name	type
items	list[str]
item	str

Returns list[str]

Effects !{}

def copied

def copied(items: list[str]) -> list[str] !{}

Parameters

name	type
items	list[str]

Returns list[str]

Effects !{}

def first_n

def first_n(items: list[str], count: int) -> list[str] !{}

Parameters

name	type
items	list[str]
count	int

Returns list[str]

Effects !{}

def subset_of

def subset_of(items: list[str], container: list[str]) -> bool !{}

Parameters

name	type
items	list[str]
container	list[str]

Returns bool

Effects !{}

def is_complete

def is_complete(completed: list[str]) -> bool !{}

Parameters

name	type
completed	list[str]

Returns bool

Effects !{}

def next_ready

def next_ready(completed: list[str]) -> str !{}

Parameters

name	type
completed	list[str]

Returns str

Effects !{}

def earliest_target

def earliest_target(targets: list[str]) -> str !{}

Parameters

name	type
targets	list[str]

Returns str

Effects !{}

def dropped_after_invalidation

def dropped_after_invalidation(frontier: str, completed: list[str]) -> list[str] !{}

Parameters

name	type
frontier	str
completed	list[str]

Returns list[str]

Effects !{}

The frontier plus its transitive downstream among the completed phases, in completed order (a candidate drops when the frontier or an already-dropped phase is among its dependencies).

def kept_after_invalidation

def kept_after_invalidation(dropped: list[str], completed: list[str]) -> list[str] !{}

Parameters

name	type
dropped	list[str]
completed	list[str]

Returns list[str]

Effects !{}

def jstr

def jstr(text: str) -> str !{}

Parameters

name	type
text	str

Returns str

Effects !{}

def jlist

def jlist(items: list[str]) -> str !{}

Parameters

name	type
items	list[str]

Returns str

Effects !{}

def jstrs

def jstrs(items: list[str]) -> str !{}

Parameters

name	type
items	list[str]

Returns str

Effects !{}

def render_execution

def render_execution(iteration: int, phase: str, segment: int, seq: int) -> str !{}

Parameters

name	type
iteration	int
phase	str
segment	int
seq	int

Returns str

Effects !{}

def render_invalidation

def render_invalidation(dropped: list[str], frontier: str, iteration: int, kept: list[str]) -> str !{}

Parameters

name	type
dropped	list[str]
frontier	str
iteration	int
kept	list[str]

Returns str

Effects !{}

def render_context

def render_context(iteration: int, prior: list[str], resume_target: str, targets: list[str]) -> str !{}

Parameters

name	type
iteration	int
prior	list[str]
resume_target	str
targets	list[str]

Returns str

Effects !{}

def render_rehydration

def render_rehydration(checkpoint: str, segment: int) -> str !{}

Parameters

name	type
checkpoint	str
segment	int

Returns str

Effects !{}

def render_settle

def render_settle(kind: str, reason: str) -> str !{}

Parameters

name	type
kind	str
reason	str

Returns str

Effects !{}

def render_trace

def render_trace(scenario: str, checkpoints: list[str], completed: list[str], executions: list[str], invalidations: list[str], contexts: list[str], loop_budget: int, loop_iterations: int, rehydrations: list[str], settle: str, status: str) -> str !{}

Parameters

name	type
scenario	str
checkpoints	list[str]
completed	list[str]
executions	list[str]
invalidations	list[str]
contexts	list[str]
loop_budget	int
loop_iterations	int
rehydrations	list[str]
settle	str
status	str

Returns str

Effects !{}

def run_scenario

def run_scenario(scenario: str, profile: str, loop_budget: int, seg1_budget: int) -> str !{model.invoke}

Parameters

name	type
scenario	str
profile	str
loop_budget	int
seg1_budget	int

Returns str

Effects !{model.invoke}

Drive one run of the 15-phase graph under the pinned engine semantics.

seg1_budget > 0 splits the drive into two segments with a process-restart boundary between them: live workspace materials die at the boundary and MUST come back from the recorded checkpoint (rehydration) for the second segment’s dependency gate to pass. Committed phases are never re-executed.

circuit run_suite

circuit run_suite() -> list[str] !{model.invoke}

Returns list[str]

Effects !{model.invoke}

def main

def main() -> str !{model.invoke}

Returns str

Effects !{model.invoke}