From 06e5c0d245e6a1676ddbaf7263045c8adff9cdb0 Mon Sep 17 00:00:00 2001 From: Allan Scott Date: Mon, 20 Apr 2026 05:16:40 -0400 Subject: [PATCH] Implement add-classical-context-updates (grammar, verifier, compiler) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ships the language/AST/verifier/compiler facets of classical context updates — the one primitive the QPC research proposal called out as missing. Grammar is: ::= | if bits[i] == v: [else: ] ::= []? (= | += | -=) ( | ) Shot-to-shot runtime execution of the mutation remains out of scope — compilers emit annotations and a banner flagging the no-op, per the spec. - AST: QContextMutation, QEffectContextUpdate, context_update field on QActionSignature. - Parser: _parse_context_update_from_effect + mixed-effect rejection + nested-conditional rejection. - Verifier: new classical_context stage wired between determinism and quantum-static; enforces UNDECLARED_CONTEXT_FIELD, CONTEXT_FIELD_TYPE_MISMATCH, CONTEXT_INDEX_OUT_OF_RANGE, and BIT_READ_BEFORE_WRITE via acyclic-path enumeration from initial. VerifyOptions.skip_classical_context added. - Compiler: QASM and Qiskit emit `// context_update: ...` / `# context_update: ...` at the action site plus a file-level banner when any context-update is present. Mermaid unchanged. - Tests: tests/test_context_updates.py (23 cases across parser, verifier, and compiler); full suite 468 passed, 5 skipped; all 10 examples still verify. Co-Authored-By: Claude Opus 4.7 --- .../research/spec-quantum-predictive-coder.md | 2 +- .../add-classical-context-updates/tasks.md | 82 ++-- q_orca/ast.py | 21 + q_orca/compiler/qasm.py | 9 + q_orca/compiler/qiskit.py | 8 + q_orca/parser/markdown_parser.py | 280 +++++++++++++ q_orca/verifier/__init__.py | 7 + q_orca/verifier/classical_context.py | 271 ++++++++++++ tests/test_context_updates.py | 395 ++++++++++++++++++ 9 files changed, 1030 insertions(+), 45 deletions(-) create mode 100644 q_orca/verifier/classical_context.py create mode 100644 tests/test_context_updates.py diff --git a/docs/research/spec-quantum-predictive-coder.md b/docs/research/spec-quantum-predictive-coder.md index 9c81cf6..e181baa 100644 --- a/docs/research/spec-quantum-predictive-coder.md +++ b/docs/research/spec-quantum-predictive-coder.md @@ -35,7 +35,7 @@ The predictive-coding loop is then a **hybrid** q-orca machine: the unitary piec - Parametric gates with `angle_context` references — **shipped** (see `openspec/changes/archive/2026-04-18-context-angle-references/`) - Mid-circuit measurement (`measure(qs[N]) -> bits[M]`) and classical-feedforward conditionals — **shipped** (archive `2026-04-17-mid-circuit-measurement`) - Runtime state-category assertions (e.g., `[assert:separable]`, `[assert:entangled]`) — **proposed** in `openspec/changes/add-runtime-state-assertions/`; would let the verifier confirm the ancilla is disentangled after measurement -- Parameter-update actions that mutate `list` context fields — **not yet shipped**; this is the one new primitive the QPC needs +- Parameter-update actions that mutate `list` context fields — **grammar, AST, verifier, and compiler annotations landed** in OpenSpec change `add-classical-context-updates`. Shot-to-shot runtime execution of the mutation is parked for a follow-up change. Nothing in this architecture requires rewriting the q-orca execution model. What it requires is a single new action kind: a **classical context update** that reads a classical bit from the machine's bit-register and writes a new float into the angle register. Concretely: `if bits[0] == 1: θ[0] -= η · δ; else: θ[0] += η · δ` as an effect, where `δ` and `η` are further context fields. This is a strictly classical operation on the context record and does not touch the quantum state; it only runs between simulator shots. diff --git a/openspec/changes/add-classical-context-updates/tasks.md b/openspec/changes/add-classical-context-updates/tasks.md index 3ecc222..34caa89 100644 --- a/openspec/changes/add-classical-context-updates/tasks.md +++ b/openspec/changes/add-classical-context-updates/tasks.md @@ -1,33 +1,33 @@ ## 1. AST -- [ ] 1.1 Add `QContextMutation` dataclass in `q_orca/ast.py` with +- [x] 1.1 Add `QContextMutation` dataclass in `q_orca/ast.py` with fields `target_field: str`, `target_idx: Optional[int]`, `op: str` (= | += | -=), `rhs_literal: Optional[float]`, `rhs_field: Optional[str]`. -- [ ] 1.2 Add `QEffectContextUpdate` dataclass with fields +- [x] 1.2 Add `QEffectContextUpdate` dataclass with fields `bit_idx: Optional[int]`, `bit_value: Optional[int]`, `then_mutations: list[QContextMutation]`, `else_mutations: list[QContextMutation]`. -- [ ] 1.3 Add `context_update: Optional[QEffectContextUpdate] = None` +- [x] 1.3 Add `context_update: Optional[QEffectContextUpdate] = None` to `QActionSignature`. ## 2. Parser -- [ ] 2.1 Add `_parse_context_update_from_effect(effect_str, errors, +- [x] 2.1 Add `_parse_context_update_from_effect(effect_str, errors, action_name)` in `q_orca/parser/markdown_parser.py`. Handle the unconditional ` ` form first, then the `if bits[i] == v: ... else: ...` form. Return `None` if the effect doesn't match the grammar (so other parsers still get a chance). -- [ ] 2.2 Wire the new parser into `_parse_actions_table` alongside +- [x] 2.2 Wire the new parser into `_parse_actions_table` alongside the existing measurement/conditional parsers. If a row produces both a `context_update` and any of `gate` / `measurement` / `conditional_gate`, emit a structured parse error and drop the row's `context_update`. -- [ ] 2.3 Enforce LHS constraints at parse time: the LHS identifier +- [x] 2.3 Enforce LHS constraints at parse time: the LHS identifier must be a simple ident (no dot paths); indexed LHS requires a non-negative integer literal index. -- [ ] 2.4 Unit tests in `tests/test_parser.py` covering: scalar +- [x] 2.4 Unit tests in `tests/test_context_updates.py` covering: scalar increment, list-element increment with literal RHS, list-element update with field-ref RHS, conditional form with then+else, conditional with only then-branch, malformed forms (unknown op, @@ -35,71 +35,65 @@ ## 3. Verifier — classical context stage -- [ ] 3.1 Create `q_orca/verifier/classical_context.py` with +- [x] 3.1 Create `q_orca/verifier/classical_context.py` with `check_classical_context(machine: QMachineDef) -> QVerificationResult`. Iterate all actions with `context_update`. -- [ ] 3.2 Implement the typing check (Requirement: +- [x] 3.2 Implement the typing check (Requirement: "Classical Context Update — Static Typing") — emit `UNDECLARED_CONTEXT_FIELD`, `CONTEXT_FIELD_TYPE_MISMATCH`, `CONTEXT_INDEX_OUT_OF_RANGE` per the spec. -- [ ] 3.3 Implement the feedforward-completeness check (Requirement: +- [x] 3.3 Implement the feedforward-completeness check (Requirement: "Classical Context Update — Feedforward Completeness") using - `analyze_machine` for reachability. Walk paths from initial - state; for each context-update with a `bit_idx`, confirm every - path to that transition contains a prior `mid_circuit_measure` - or `measurement` writing that bit. Emit + acyclic-path enumeration from the initial state. For each + context-update with a `bit_idx`, confirm every path to that + transition contains a prior `mid_circuit_measure` or + `measurement` writing that bit. Emit `BIT_READ_BEFORE_WRITE` on violation. -- [ ] 3.4 Wire the stage into `q_orca/verifier/__init__.py::verify()` - between completeness and quantum-static. Respect - `VerifyOptions.skip_classical_context` (add that flag to - `VerifyOptions`). -- [ ] 3.5 Unit tests in `tests/test_verifier.py` covering each +- [x] 3.4 Wire the stage into `q_orca/verifier/__init__.py::verify()` + between completeness/determinism and quantum-static. Respect + `VerifyOptions.skip_classical_context` (added to `VerifyOptions`). +- [x] 3.5 Unit tests in `tests/test_context_updates.py` covering each error code (missing field, wrong type, out-of-range index, bit-read-before-write) and the happy path where a measurement transition writes the bit before the update. ## 4. Compiler — annotation emission -- [ ] 4.1 In `q_orca/compiler/qasm.py`, detect +- [x] 4.1 In `q_orca/compiler/qasm.py`, detect `QEffectContextUpdate` actions and emit a `// context_update: ...` comment at the action's site. Track presence-of-any to decide whether to emit the file-level banner. -- [ ] 4.2 Same treatment in `q_orca/compiler/qiskit.py` with Python +- [x] 4.2 Same treatment in `q_orca/compiler/qiskit.py` with Python `#` comments. -- [ ] 4.3 In `q_orca/compiler/mermaid.py`, confirm transition-arrow - labels render unchanged — likely no code change, just a test. -- [ ] 4.4 Serialize the original effect string on - `QEffectContextUpdate` at parse time (add `raw: str` field) so +- [x] 4.3 In `q_orca/compiler/mermaid.py`, confirm transition-arrow + labels render unchanged — no code change needed; covered by test. +- [x] 4.4 Serialize the original effect string on + `QEffectContextUpdate` at parse time (`raw: str` field) so compilers can emit the round-trippable text without re-stringifying the AST. -- [ ] 4.5 Unit tests in `tests/test_compiler.py` covering QASM and - Qiskit emission (presence of comment, presence of banner) and +- [x] 4.5 Unit tests in `tests/test_context_updates.py` covering QASM + and Qiskit emission (presence of comment, presence of banner) and the no-context-update negative case (banner absent). ## 5. Spec + docs sync -- [ ] 5.1 Run `openspec validate add-classical-context-updates - --strict` — already required at the end of the OpenSpec propose - flow; re-run after code lands. -- [ ] 5.2 Add a one-line pointer in - `docs/research/spec-quantum-predictive-coder.md` referencing this - change ID as the blocker the research proposal flagged (research - doc's §Proposed Architecture item 5, "Parameter-update - actions...not yet shipped"). +- [x] 5.1 Run `openspec validate add-classical-context-updates + --strict` — performed after code lands. +- [x] 5.2 Update `docs/research/spec-quantum-predictive-coder.md` + to reference this change as the source of the grammar/AST/verifier/ + compiler landing (§Proposed Architecture item 5). ## 6. End-to-end verification -- [ ] 6.1 Write a minimal fixture machine using the new grammar - (extends `examples/predictive-coder-minimal.q.orca.md` with a - `gradient_step` action and a loop-back transition). Confirm it - parses, verifies, and compiles to QASM/Qiskit with annotations - present. -- [ ] 6.2 Run the full suite +- [x] 6.1 Inline fixture machine in `tests/test_context_updates.py` + exercises the new grammar through parser, verifier, and both + compilers (QASM + Qiskit) end-to-end. +- [x] 6.2 Run the full suite `.venv/bin/python -m pytest tests/ -q --ignore=tests/test_cuquantum_backend.py - --ignore=tests/test_cudaq_backend.py` and confirm green. -- [ ] 6.3 Run `.venv/bin/q-orca verify` on all examples in - `examples/` and confirm none regress. + --ignore=tests/test_cudaq_backend.py` — 468 passed, 5 skipped. +- [x] 6.3 Run `.venv/bin/q-orca verify` on all examples in + `examples/` — all 10 remain VALID. ## 7. Follow-up parked (NOT this change) diff --git a/q_orca/ast.py b/q_orca/ast.py index 9fb4b32..44e0b6d 100644 --- a/q_orca/ast.py +++ b/q_orca/ast.py @@ -209,6 +209,26 @@ class QEffectConditional: gate: QuantumGate +@dataclass +class QContextMutation: + """A single classical-context mutation: .""" + target_field: str + target_idx: Optional[int] = None # None for scalar, int for list element + op: str = "=" # "=", "+=", "-=" + rhs_literal: Optional[float] = None + rhs_field: Optional[str] = None # mutually exclusive with rhs_literal + + +@dataclass +class QEffectContextUpdate: + """Optionally-bit-gated mutation of classical context fields.""" + then_mutations: list["QContextMutation"] = field(default_factory=list) + else_mutations: list["QContextMutation"] = field(default_factory=list) + bit_idx: Optional[int] = None + bit_value: Optional[int] = None # 0 or 1; None iff bit_idx is None + raw: Optional[str] = None # original effect string for round-trip emission + + @dataclass class QActionSignature: name: str @@ -221,6 +241,7 @@ class QActionSignature: measurement: Optional[Measurement] = None mid_circuit_measure: Optional[QEffectMeasure] = None conditional_gate: Optional[QEffectConditional] = None + context_update: Optional[QEffectContextUpdate] = None @dataclass diff --git a/q_orca/compiler/qasm.py b/q_orca/compiler/qasm.py index 2c55441..f7c2e45 100644 --- a/q_orca/compiler/qasm.py +++ b/q_orca/compiler/qasm.py @@ -11,6 +11,12 @@ def compile_to_qasm(machine: QMachineDef) -> str: lines.append("// Generated by Q-Orca compiler") lines.append(f"// Machine: {machine.name}") + has_context_update = any(a.context_update is not None for a in machine.actions) + if has_context_update: + lines.append( + "// NOTE: context-update actions are annotations only; " + "shot-to-shot execution not yet implemented." + ) lines.append('OPENQASM 3.0;') lines.append('include "stdgates.inc";') lines.append("") @@ -55,6 +61,9 @@ def compile_to_qasm(machine: QMachineDef) -> str: # OpenQASM 3.0 per-bit conditional: bare bit for 1, negated for 0 cond = f"c[{cg.bit_idx}]" if cg.value else f"!c[{cg.bit_idx}]" lines.append(f"if ({cond}) {{ {gate_str}; }}") + elif action and action.context_update is not None: + raw = action.context_update.raw or action.effect or "" + lines.append(f"// context_update: {raw}") else: for gate in gates: lines.append(_gate_to_qasm(gate, qubit_count)) diff --git a/q_orca/compiler/qiskit.py b/q_orca/compiler/qiskit.py index d0cec3f..dc8393d 100644 --- a/q_orca/compiler/qiskit.py +++ b/q_orca/compiler/qiskit.py @@ -276,6 +276,11 @@ def compile_to_qiskit(machine: QMachineDef, options: QSimulationOptions) -> str: lines.append("# Generated by Q-Orca compiler") lines.append(f"# Machine: {machine.name}") + if any(a.context_update is not None for a in machine.actions): + lines.append( + "# NOTE: context-update actions are annotations only; " + "shot-to-shot execution not yet implemented." + ) lines.append("") lines.append("import json") @@ -338,6 +343,9 @@ def compile_to_qiskit(machine: QMachineDef, options: QSimulationOptions) -> str: cg = action.conditional_gate lines.append(f"with qc.if_test((qc.clbits[{cg.bit_idx}], {cg.value})):") lines.append(f" {_gate_to_qiskit(cg.gate)}") + elif action and action.context_update is not None: + raw = action.context_update.raw or action.effect or "" + lines.append(f"# context_update: {raw}") elif gates: for gate in gates: lines.append(_gate_to_qiskit(gate)) diff --git a/q_orca/parser/markdown_parser.py b/q_orca/parser/markdown_parser.py index fabe6b0..f48b705 100644 --- a/q_orca/parser/markdown_parser.py +++ b/q_orca/parser/markdown_parser.py @@ -14,6 +14,7 @@ QGuardRef, QuantumGate, Measurement, CollapseOutcome, QGuardTrue, QGuardFalse, QGuardCompare, QGuardProbability, QGuardFidelity, VariableRef, ValueRef, QEffectMeasure, QEffectConditional, + QContextMutation, QEffectContextUpdate, ) @@ -467,17 +468,51 @@ def _parse_actions_table( continue params, return_type = _parse_signature(sig_str) + context_update = _parse_context_update_from_effect(effect_str, errors, action_name=name) gate = _parse_gate_from_effect(effect_str, errors, action_name=name, angle_context=angle_context) measurement = _parse_measurement_from_effect(effect_str) mid_circuit_measure = _parse_mid_circuit_measure_from_effect(effect_str) conditional_gate = _parse_conditional_gate_from_effect(effect_str, errors, action_name=name, angle_context=angle_context) + has_other_effect = ( + gate is not None + or measurement is not None + or mid_circuit_measure is not None + or conditional_gate is not None + ) + + # If the parser recognized a context-update AND any other effect, + # that's a mixed-kind effect — rejected in v1. + if context_update is not None and has_other_effect: + if errors is not None: + errors.append( + f"action {name!r}: context-update effect cannot be combined with " + f"gate, measurement, mid-circuit measurement, or conditional-gate " + f"effects in a single action (v1)." + ) + context_update = None + # Otherwise, a gate+something string may still hide an unparsed + # mutation tail (e.g. `H(qs[0]); iteration += 1`). Detect the + # mutation-operator pattern in any `;`-delimited segment that isn't + # the first and isn't a parsed gate. + elif has_other_effect and effect_str and _has_trailing_mutation(effect_str): + if errors is not None: + errors.append( + f"action {name!r}: context-update effect cannot be combined with " + f"gate, measurement, mid-circuit measurement, or conditional-gate " + f"effects in a single action (v1)." + ) + + # Skip the "looks-like-gate" warning when the effect parsed as a + # context-update; otherwise we'd spuriously flag `iteration += 1` + # as a gate typo. if ( effect_str and gate is None and measurement is None and mid_circuit_measure is None and conditional_gate is None + and context_update is None and errors is not None and _looks_like_gate_call(effect_str) # Don't double-fire if _parse_gate_from_effect already surfaced a @@ -501,6 +536,7 @@ def _parse_actions_table( measurement=measurement, mid_circuit_measure=mid_circuit_measure, conditional_gate=conditional_gate, + context_update=context_update, )) return actions @@ -909,3 +945,247 @@ def _parse_conditional_gate_from_effect( if gate is None: return None return QEffectConditional(bit_idx=bit_idx, value=value, gate=gate) + + +# ============================================================ +# Context-update parsing +# ============================================================ + +# A single mutation like: iteration += 1 | theta[0] -= eta +_MUTATION_RE = re.compile( + r""" + ^\s* + (?P[A-Za-z_][A-Za-z0-9_]*) # field name + (?:\[\s*(?P-?\d+)\s*\])? # optional [int] index + \s*(?P=|\+=|-=)\s* + (?P[^\s].*?) + \s*$ + """, + re.VERBOSE, +) + +_FLOAT_LITERAL_RE = re.compile(r"^-?\d+(\.\d+)?([eE][+-]?\d+)?$") + + +def _parse_single_mutation( + mut_str: str, + errors: list[str] | None, + action_name: str, +) -> Optional[QContextMutation]: + """Parse one ` ` atom into a QContextMutation.""" + m = _MUTATION_RE.match(mut_str) + if not m: + if errors is not None: + errors.append( + f"action {action_name!r}: malformed context-update mutation {mut_str!r} " + f"(expected ` = | += | -= `)." + ) + return None + + target_field = m.group("lhs") + idx_str = m.group("idx") + target_idx: Optional[int] = None + if idx_str is not None: + try: + target_idx = int(idx_str) + except ValueError: + if errors is not None: + errors.append( + f"action {action_name!r}: non-integer list index in mutation {mut_str!r}." + ) + return None + if target_idx < 0: + if errors is not None: + errors.append( + f"action {action_name!r}: negative list index in mutation {mut_str!r}." + ) + return None + + op = m.group("op") + rhs = m.group("rhs").strip() + + rhs_literal: Optional[float] = None + rhs_field: Optional[str] = None + if _FLOAT_LITERAL_RE.match(rhs): + try: + rhs_literal = float(rhs) + except ValueError: + if errors is not None: + errors.append( + f"action {action_name!r}: unrecognized numeric RHS {rhs!r} in mutation {mut_str!r}." + ) + return None + elif re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", rhs): + rhs_field = rhs + else: + if errors is not None: + errors.append( + f"action {action_name!r}: RHS {rhs!r} in mutation {mut_str!r} must be a " + f"numeric literal or a bare context-field identifier." + ) + return None + + return QContextMutation( + target_field=target_field, + target_idx=target_idx, + op=op, + rhs_literal=rhs_literal, + rhs_field=rhs_field, + ) + + +def _parse_mutation_sequence( + seq_str: str, + errors: list[str] | None, + action_name: str, +) -> list[QContextMutation]: + """Parse `mut (; mut)*` into a list of QContextMutation.""" + mutations: list[QContextMutation] = [] + for piece in seq_str.split(";"): + piece = piece.strip() + if not piece: + continue + mut = _parse_single_mutation(piece, errors, action_name) + if mut is not None: + mutations.append(mut) + return mutations + + +def _parse_context_update_from_effect( + effect_str: str, + errors: list[str] | None = None, + action_name: str = "", +) -> Optional[QEffectContextUpdate]: + """Parse a context-update effect string into a QEffectContextUpdate. + + Returns None if the effect does not match the context-update grammar, + so the other effect parsers still get a chance. Emits structured + errors only for forms that are clearly context-update intent but + malformed. + """ + if not effect_str: + return None + + stripped = effect_str.strip() + + # Ignore the conditional-gate form: it starts with `if bits[...` but the + # body is a gate call, not a mutation. We detect "context-update intent" + # by the presence of `;` or a mutation operator (= / += / -=) somewhere + # after the colon. + cond_match = re.match( + r"^if\s+bits\[(\d+)\]\s*==\s*([01])\s*:\s*(.*)$", + stripped, + re.IGNORECASE, + ) + if cond_match: + bit_idx = int(cond_match.group(1)) + bit_value = int(cond_match.group(2)) + body = cond_match.group(3).strip() + then_part, else_part = _split_then_else(body) + + has_nested_if = bool( + re.search(r"\bif\s+bits\[", then_part, re.IGNORECASE) + or (else_part is not None and re.search(r"\bif\s+bits\[", else_part, re.IGNORECASE)) + ) + + # If the body contains nested `if bits[...]`, decide whether the + # innermost form is mutation-intent (context-update) or gate-intent + # (nested conditional gates — also not supported, but handled + # elsewhere). Look anywhere in the body for a mutation operator + # (`=` / `+=` / `-=`, not `==`) on a field-like LHS; if present, + # the user is writing a nested context-update. + if has_nested_if: + if re.search(r"[A-Za-z_]\w*(?:\[\s*-?\d+\s*\])?\s*(\+=|-=|(? bool: + """Heuristic: text begins with `([])? ` where op is =/+=/-=. + + The op must not be `==` (that's a comparison, used in bit conditions). + """ + if not text: + return False + m = re.match( + r"^\s*[A-Za-z_][A-Za-z0-9_]*(?:\[\s*-?\d+\s*\])?\s*(=(?!=)|\+=|-=)", + text, + ) + return m is not None + + +# Any `([])? (= | += | -=)` occurring after a semicolon or at a +# non-start position — used to detect mixed gate/context-update effects +# like `H(qs[0]); iteration += 1`. +_MUTATION_OP_PAT = re.compile( + r"(?:^|;)\s*[A-Za-z_][A-Za-z0-9_]*(?:\[\s*-?\d+\s*\])?\s*(=(?!=)|\+=|-=)" +) + + +def _has_trailing_mutation(effect_str: str) -> bool: + """True if any segment of `effect_str` looks like a mutation. + + Used to catch `gate; mutation` combinations that the context-update + parser rejected as a whole but that still indicate mixed intent. + """ + return _MUTATION_OP_PAT.search(effect_str) is not None + + +def _split_then_else(body: str) -> tuple[str, Optional[str]]: + """Split a context-update body on a top-level `else:` keyword. + + Only splits on `else:` that appears between semicolon-separated + mutations — otherwise we'd mis-split on an `else` inside some other + construct. For v1 (no nesting), a simple `else:` token search is + sufficient. + """ + # Look for `else:` with word boundaries; take the first match. + m = re.search(r"\belse\s*:\s*", body) + if not m: + return body.strip(), None + then_part = body[:m.start()].strip().rstrip(";").strip() + else_part = body[m.end():].strip() + return then_part, else_part diff --git a/q_orca/verifier/__init__.py b/q_orca/verifier/__init__.py index 8660feb..053ac23 100644 --- a/q_orca/verifier/__init__.py +++ b/q_orca/verifier/__init__.py @@ -7,6 +7,7 @@ from q_orca.verifier.structural import check_structural from q_orca.verifier.completeness import check_completeness from q_orca.verifier.determinism import check_determinism +from q_orca.verifier.classical_context import check_classical_context from q_orca.verifier.quantum import verify_quantum from q_orca.verifier.superposition import check_superposition_leaks from q_orca.verifier.types import QVerificationResult, QVerificationError @@ -18,6 +19,7 @@ class VerifyOptions: skip_quantum: bool = False skip_qutip: bool = False skip_dynamic: bool = False + skip_classical_context: bool = False backend: str = "qutip" @@ -42,6 +44,11 @@ def verify(machine: QMachineDef, options: Optional[VerifyOptions] = None) -> QVe determinism = check_determinism(machine) all_errors.extend(determinism.errors) + # Stage 3b: Classical-context (types + feedforward completeness) + if not opts.skip_classical_context: + classical = check_classical_context(machine) + all_errors.extend(classical.errors) + # Stage 4: Quantum-specific checks if not opts.skip_quantum: quantum = verify_quantum(machine) diff --git a/q_orca/verifier/classical_context.py b/q_orca/verifier/classical_context.py new file mode 100644 index 0000000..8a93c1c --- /dev/null +++ b/q_orca/verifier/classical_context.py @@ -0,0 +1,271 @@ +"""Q-Orca classical-context verification. + +Two static checks on context-update actions: + +1. Static typing — the LHS of every mutation must reference a declared + context field of the right kind, and list-index LHSs must be within + the field's default-value bounds. RHS field refs must also be + declared numeric fields. + +2. Feedforward completeness — any context-update effect that reads + `bits[i]` in its condition must be preceded, on every reachable + path from the initial state, by a transition that writes + `bits[i]` (via `measure(qs[_]) -> bits[i]`). +""" + +import re + +from q_orca.ast import ( + QMachineDef, + QActionSignature, + QContextMutation, + ContextField, + QTypeScalar, + QTypeList, +) +from q_orca.verifier.types import QVerificationError, QVerificationResult + + +_NUMERIC_SCALARS = {"int", "float"} + + +def _context_field_by_name(machine: QMachineDef, name: str) -> ContextField | None: + for f in machine.context: + if f.name == name: + return f + return None + + +def _list_default_length(default_value: str | None) -> int | None: + if not default_value: + return None + inner = default_value.strip() + if inner.startswith("[") and inner.endswith("]"): + inner = inner[1:-1].strip() + if not inner: + return 0 + # Split on commas at the top level (defaults in this grammar are + # flat numeric lists — no nested brackets to worry about). + return len([p for p in inner.split(",") if p.strip()]) + + +def _check_mutation_typing( + mut: QContextMutation, + machine: QMachineDef, + action_name: str, +) -> list[QVerificationError]: + errors: list[QVerificationError] = [] + + field = _context_field_by_name(machine, mut.target_field) + if field is None: + errors.append(QVerificationError( + code="UNDECLARED_CONTEXT_FIELD", + message=( + f"Action '{action_name}' mutates undeclared context field " + f"'{mut.target_field}'" + ), + severity="error", + location={"action": action_name, "field": mut.target_field}, + suggestion=f"Declare '{mut.target_field}' in the ## context table.", + )) + else: + # Typing: scalar LHS must be `int`; list-element LHS must be `list`. + if mut.target_idx is None: + if not (isinstance(field.type, QTypeScalar) and field.type.kind == "int"): + errors.append(QVerificationError( + code="CONTEXT_FIELD_TYPE_MISMATCH", + message=( + f"Action '{action_name}': scalar context mutation " + f"requires an `int` field, but '{mut.target_field}' " + f"has a different type." + ), + severity="error", + location={"action": action_name, "field": mut.target_field}, + )) + else: + if not ( + isinstance(field.type, QTypeList) + and field.type.element_type.strip() == "float" + ): + errors.append(QVerificationError( + code="CONTEXT_FIELD_TYPE_MISMATCH", + message=( + f"Action '{action_name}': indexed context mutation " + f"requires a `list` field, but '{mut.target_field}' " + f"has a different type." + ), + severity="error", + location={"action": action_name, "field": mut.target_field}, + )) + else: + length = _list_default_length(field.default_value) + if length is not None and mut.target_idx >= length: + errors.append(QVerificationError( + code="CONTEXT_INDEX_OUT_OF_RANGE", + message=( + f"Action '{action_name}': index {mut.target_idx} is " + f"outside the default-value length ({length}) of " + f"list field '{mut.target_field}'." + ), + severity="error", + location={"action": action_name, "field": mut.target_field}, + )) + + # RHS field refs: must exist and be numeric scalar. + if mut.rhs_field is not None: + rhs = _context_field_by_name(machine, mut.rhs_field) + if rhs is None: + errors.append(QVerificationError( + code="UNDECLARED_CONTEXT_FIELD", + message=( + f"Action '{action_name}' references undeclared context " + f"field '{mut.rhs_field}' as a mutation RHS." + ), + severity="error", + location={"action": action_name, "field": mut.rhs_field}, + )) + elif not (isinstance(rhs.type, QTypeScalar) and rhs.type.kind in _NUMERIC_SCALARS): + errors.append(QVerificationError( + code="CONTEXT_FIELD_TYPE_MISMATCH", + message=( + f"Action '{action_name}': RHS field '{mut.rhs_field}' must " + f"be an `int` or `float` scalar." + ), + severity="error", + location={"action": action_name, "field": mut.rhs_field}, + )) + + return errors + + +def _action_writes_bit(action: QActionSignature, bit_idx: int) -> bool: + """True if this action writes `bits[bit_idx]` via a measurement effect.""" + if action.mid_circuit_measure is not None and action.mid_circuit_measure.bit_idx == bit_idx: + return True + # Also honor any effect string that declares `measure(qs[_]) -> bits[i]` + # — `mid_circuit_measure` should be set in this case, but we double-check + # here to stay robust against parser changes. + if action.effect: + for m in re.finditer( + r"measure\s*\(\s*\w+\[\d+\]\s*\)\s*->\s*bits\[(\d+)\]", + action.effect, + re.IGNORECASE, + ): + if int(m.group(1)) == bit_idx: + return True + return False + + +def _check_feedforward_completeness( + machine: QMachineDef, +) -> list[QVerificationError]: + """For each bit-gated context-update, confirm every path to its + transition writes that bit first. Emit BIT_READ_BEFORE_WRITE on + any path missing the write. + """ + errors: list[QVerificationError] = [] + action_map = {a.name: a for a in machine.actions} + + initial = next((s for s in machine.states if s.is_initial), None) + if initial is None: + return errors # structural stage will flag this. + + outgoing: dict[str, list] = {} + for t in machine.transitions: + outgoing.setdefault(t.source, []).append(t) + + def _transition_action(t) -> QActionSignature | None: + return action_map.get(t.action) if t.action else None + + already_reported: set[tuple[str, int]] = set() + + def _enumerate_paths_to(target_state: str) -> list[list]: + """Return every acyclic list-of-transitions from initial to target_state. + + The empty list represents the case where initial == target_state + (i.e., the update fires on the first transition out of initial). + """ + results: list[list] = [] + + def dfs(state: str, visited_states: set, transitions_acc: list) -> None: + if state == target_state: + results.append(list(transitions_acc)) + # Don't return — the same state may be reachable through + # multiple acyclic prefixes; but *on this branch* we stop + # because continuing past would revisit. + return + for out in outgoing.get(state, []): + if out.target in visited_states: + continue + dfs( + out.target, + visited_states | {out.target}, + transitions_acc + [out], + ) + + dfs(initial.name, {initial.name}, []) + return results + + for t_target in machine.transitions: + act = _transition_action(t_target) + if act is None or act.context_update is None: + continue + cu = act.context_update + if cu.bit_idx is None: + continue + bit_idx = cu.bit_idx + + paths = _enumerate_paths_to(t_target.source) + violation = False + if not paths: + # The transition's source isn't reachable from initial; the + # structural stage already flags unreachable states, so skip. + continue + + for path_transitions in paths: + writes_bit = any( + _transition_action(tp) is not None + and _action_writes_bit(_transition_action(tp), bit_idx) + for tp in path_transitions + ) + if not writes_bit: + violation = True + break + + if violation and (act.name, bit_idx) not in already_reported: + errors.append(QVerificationError( + code="BIT_READ_BEFORE_WRITE", + message=( + f"Action '{act.name}' reads bits[{bit_idx}] in a " + f"context-update condition, but some path from the " + f"initial state reaches it without a prior " + f"measure(...) -> bits[{bit_idx}]." + ), + severity="error", + location={"action": act.name, "bit": bit_idx}, + suggestion=( + f"Add a mid-circuit measurement writing bits[{bit_idx}] " + f"on every path leading to this action." + ), + )) + already_reported.add((act.name, bit_idx)) + + return errors + + +def check_classical_context(machine: QMachineDef) -> QVerificationResult: + errors: list[QVerificationError] = [] + + for action in machine.actions: + cu = action.context_update + if cu is None: + continue + for mut in list(cu.then_mutations) + list(cu.else_mutations): + errors.extend(_check_mutation_typing(mut, machine, action.name)) + + errors.extend(_check_feedforward_completeness(machine)) + + return QVerificationResult( + valid=not any(e.severity == "error" for e in errors), + errors=errors, + ) diff --git a/tests/test_context_updates.py b/tests/test_context_updates.py new file mode 100644 index 0000000..b6a82a2 --- /dev/null +++ b/tests/test_context_updates.py @@ -0,0 +1,395 @@ +"""Tests for classical context-update effects. + +Covers the parser, verifier, and compiler facets of the +`add-classical-context-updates` change. +""" + +import pytest + +from q_orca.ast import QEffectContextUpdate +from q_orca.compiler.qasm import compile_to_qasm +from q_orca.compiler.qiskit import QSimulationOptions, compile_to_qiskit +from q_orca.parser.markdown_parser import parse_q_orca_markdown +from q_orca.verifier import verify, VerifyOptions +from q_orca.verifier.classical_context import check_classical_context + + +def _parse(source: str): + return parse_q_orca_markdown(source) + + +def _machine(source: str): + return _parse(source).file.machines[0] + + +def _base_machine_with_update(effect: str, action_name: str = "gradient_step") -> str: + """Minimal machine with a single context-update action.""" + return f"""\ +# machine WithUpdate + +## context +| Field | Type | Default | +|-------|------|---------| +| iteration | int | 0 | +| theta | list | [0.0, 0.0] | +| eta | float | 0.1 | +| qubits | list | [q0] | + +## events +- measure_out +- {action_name}_ev + +## state |ψ0> [initial] +> start + +## state |ψ1> +> after measurement + +## state |ψ2> [final] +> after update + +## transitions +| Source | Event | Guard | Target | Action | +|--------|-----------------|-------|--------|---------------| +| |ψ0> | measure_out | | |ψ1> | measure_ancilla | +| |ψ1> | {action_name}_ev | | |ψ2> | {action_name} | + +## actions +| Name | Signature | Effect | +|-----------------|------------------------|----------------------------------| +| measure_ancilla | (qs, bits) -> (qs, bits) | measure(qs[0]) -> bits[0] | +| {action_name} | (ctx) -> ctx | {effect} | +""" + + +# ============================================================ +# Parser tests +# ============================================================ + +class TestContextUpdateParser: + def test_scalar_increment(self): + source = _base_machine_with_update("iteration += 1", action_name="tick") + result = _parse(source) + machine = result.file.machines[0] + tick = next(a for a in machine.actions if a.name == "tick") + assert tick.context_update is not None + cu = tick.context_update + assert cu.bit_idx is None + assert cu.bit_value is None + assert len(cu.then_mutations) == 1 + m = cu.then_mutations[0] + assert m.target_field == "iteration" + assert m.target_idx is None + assert m.op == "+=" + assert m.rhs_literal == 1.0 + assert m.rhs_field is None + + def test_list_element_literal_rhs(self): + source = _base_machine_with_update("theta[0] -= 0.1") + machine = _parse(source).file.machines[0] + action = next(a for a in machine.actions if a.name == "gradient_step") + cu = action.context_update + assert cu is not None + assert cu.bit_idx is None + assert len(cu.then_mutations) == 1 + m = cu.then_mutations[0] + assert m.target_field == "theta" + assert m.target_idx == 0 + assert m.op == "-=" + assert m.rhs_literal == 0.1 + + def test_list_element_field_rhs(self): + source = _base_machine_with_update("theta[1] += eta") + machine = _parse(source).file.machines[0] + action = next(a for a in machine.actions if a.name == "gradient_step") + m = action.context_update.then_mutations[0] + assert m.target_field == "theta" + assert m.target_idx == 1 + assert m.op == "+=" + assert m.rhs_field == "eta" + assert m.rhs_literal is None + + def test_conditional_with_then_and_else(self): + effect = "if bits[0] == 1: theta[0] -= eta else: theta[0] += eta" + source = _base_machine_with_update(effect) + machine = _parse(source).file.machines[0] + action = next(a for a in machine.actions if a.name == "gradient_step") + cu = action.context_update + assert cu is not None + assert cu.bit_idx == 0 + assert cu.bit_value == 1 + assert len(cu.then_mutations) == 1 + assert len(cu.else_mutations) == 1 + assert cu.then_mutations[0].op == "-=" + assert cu.else_mutations[0].op == "+=" + + def test_conditional_then_only(self): + effect = "if bits[0] == 1: theta[0] -= eta" + source = _base_machine_with_update(effect) + machine = _parse(source).file.machines[0] + action = next(a for a in machine.actions if a.name == "gradient_step") + cu = action.context_update + assert cu is not None + assert cu.bit_idx == 0 + assert cu.bit_value == 1 + assert len(cu.then_mutations) == 1 + assert cu.else_mutations == [] + + def test_unconditional_multi_mutation(self): + effect = "theta[0] += eta; iteration += 1" + source = _base_machine_with_update(effect, action_name="step") + machine = _parse(source).file.machines[0] + action = next(a for a in machine.actions if a.name == "step") + cu = action.context_update + assert cu is not None + assert len(cu.then_mutations) == 2 + assert cu.then_mutations[0].target_field == "theta" + assert cu.then_mutations[1].target_field == "iteration" + + def test_mixed_gate_and_context_update_rejected(self): + effect = "H(qs[0]); iteration += 1" + source = _base_machine_with_update(effect, action_name="bad") + result = _parse(source) + assert any("cannot be combined" in e for e in result.errors), result.errors + machine = result.file.machines[0] + action = next(a for a in machine.actions if a.name == "bad") + assert action.context_update is None + + def test_nested_conditional_rejected(self): + effect = "if bits[0] == 1: if bits[1] == 1: theta[0] -= eta" + source = _base_machine_with_update(effect, action_name="bad") + result = _parse(source) + assert any("nested" in e.lower() for e in result.errors), result.errors + + def test_non_bit_condition_not_parsed_as_update(self): + # `if iteration > 0: ...` isn't valid grammar; the parser should + # return None (letting other parsers see a non-match) rather than + # producing a context_update. + effect = "if iteration > 0: theta[0] += eta" + source = _base_machine_with_update(effect, action_name="bad") + machine = _parse(source).file.machines[0] + action = next(a for a in machine.actions if a.name == "bad") + assert action.context_update is None + + def test_raw_effect_string_preserved(self): + effect = "if bits[0] == 1: theta[0] -= eta else: theta[0] += eta" + source = _base_machine_with_update(effect) + machine = _parse(source).file.machines[0] + action = next(a for a in machine.actions if a.name == "gradient_step") + assert action.context_update.raw == effect + + +# ============================================================ +# Verifier tests +# ============================================================ + +class TestContextUpdateVerifier: + def test_happy_path_verifies_cleanly(self): + effect = "if bits[0] == 1: theta[0] -= eta else: theta[0] += eta" + source = _base_machine_with_update(effect) + machine = _machine(source) + res = check_classical_context(machine) + assert res.valid, [e.code for e in res.errors] + + def test_undeclared_field_reports_error(self): + source = _base_machine_with_update("nonexistent += 1", action_name="tick") + machine = _machine(source) + res = check_classical_context(machine) + codes = [e.code for e in res.errors] + assert "UNDECLARED_CONTEXT_FIELD" in codes + + def test_wrong_type_scalar_mutation(self): + source = f"""\ +# machine WrongType + +## context +| Field | Type | Default | +|-------|------|---------| +| label | string | "foo" | +| qubits | list | [q0] | + +## events +- tick_ev + +## state |ψ0> [initial] + +## state |ψ1> [final] + +## transitions +| Source | Event | Guard | Target | Action | +|--------|---------|-------|--------|--------| +| |ψ0> | tick_ev | | |ψ1> | tick | + +## actions +| Name | Signature | Effect | +|------|---------------|-------------| +| tick | (ctx) -> ctx | label += 1 | +""" + machine = _machine(source) + res = check_classical_context(machine) + codes = [e.code for e in res.errors] + assert "CONTEXT_FIELD_TYPE_MISMATCH" in codes + + def test_index_out_of_range(self): + source = _base_machine_with_update("theta[5] += 0.1", action_name="tick") + machine = _machine(source) + res = check_classical_context(machine) + codes = [e.code for e in res.errors] + assert "CONTEXT_INDEX_OUT_OF_RANGE" in codes + + def test_bit_read_before_write_no_measurement_anywhere(self): + source = f"""\ +# machine Unread + +## context +| Field | Type | Default | +|-------|------|---------| +| theta | list | [0.0] | +| eta | float | 0.1 | +| qubits | list | [q0] | + +## events +- go + +## state |ψ0> [initial] + +## state |ψ1> [final] + +## transitions +| Source | Event | Guard | Target | Action | +|--------|-------|-------|--------|---------------| +| |ψ0> | go | | |ψ1> | gradient_step | + +## actions +| Name | Signature | Effect | +|----------------|--------------|------------------------------------------------------| +| gradient_step | (ctx) -> ctx | if bits[0] == 1: theta[0] -= eta else: theta[0] += eta | +""" + machine = _machine(source) + res = check_classical_context(machine) + codes = [e.code for e in res.errors] + assert "BIT_READ_BEFORE_WRITE" in codes + + def test_bit_written_on_every_path_is_ok(self): + effect = "if bits[0] == 1: theta[0] -= eta" + source = _base_machine_with_update(effect) + machine = _machine(source) + res = check_classical_context(machine) + codes = [e.code for e in res.errors] + assert "BIT_READ_BEFORE_WRITE" not in codes + + def test_skip_flag_disables_stage(self): + source = _base_machine_with_update("nonexistent += 1", action_name="tick") + machine = _machine(source) + # Normally this would fail classical-context + res_skip = verify(machine, VerifyOptions(skip_classical_context=True, + skip_dynamic=True, + skip_quantum=True)) + codes = [e.code for e in res_skip.errors] + assert "UNDECLARED_CONTEXT_FIELD" not in codes + + def test_pipeline_integration(self): + """Classical-context errors surface through the top-level verify().""" + source = _base_machine_with_update("nonexistent += 1", action_name="tick") + machine = _machine(source) + res = verify(machine, VerifyOptions(skip_dynamic=True, skip_quantum=True)) + codes = [e.code for e in res.errors] + assert "UNDECLARED_CONTEXT_FIELD" in codes + assert not res.valid + + +# ============================================================ +# Compiler tests +# ============================================================ + +class TestContextUpdateCompiler: + def test_qasm_emits_annotation_and_banner(self): + effect = "if bits[0] == 1: theta[0] -= eta else: theta[0] += eta" + source = _base_machine_with_update(effect) + machine = _machine(source) + qasm = compile_to_qasm(machine) + assert f"// context_update: {effect}" in qasm + assert "context-update actions are annotations only" in qasm + + def test_qasm_no_banner_when_no_context_update(self): + # Minimal machine with only a gate action. + source = """\ +# machine NoUpdate + +## context +| Field | Type | Default | +|-------|------|---------| +| qubits | list | [q0] | + +## events +- go + +## state |0> [initial] + +## state |1> [final] + +## transitions +| Source | Event | Guard | Target | Action | +|--------|-------|-------|--------|--------| +| |0> | go | | |1> | apply_x | + +## actions +| Name | Signature | Effect | +|--------|------------------|-----------| +| apply_x | (qs) -> qs | X(qs[0]) | +""" + machine = _machine(source) + qasm = compile_to_qasm(machine) + assert "context-update actions are annotations only" not in qasm + assert "context_update" not in qasm + + def test_qiskit_emits_annotation_and_banner(self): + effect = "iteration += 1" + source = _base_machine_with_update(effect, action_name="tick") + machine = _machine(source) + qiskit_script = compile_to_qiskit(machine, QSimulationOptions(skip_qutip=True)) + assert f"# context_update: {effect}" in qiskit_script + assert "context-update actions are annotations only" in qiskit_script + + def test_qiskit_no_banner_when_no_context_update(self): + source = """\ +# machine NoUpdate + +## context +| Field | Type | Default | +|-------|------|---------| +| qubits | list | [q0] | + +## events +- go + +## state |0> [initial] + +## state |1> [final] + +## transitions +| Source | Event | Guard | Target | Action | +|--------|-------|-------|--------|--------| +| |0> | go | | |1> | apply_x | + +## actions +| Name | Signature | Effect | +|--------|------------------|-----------| +| apply_x | (qs) -> qs | X(qs[0]) | +""" + machine = _machine(source) + script = compile_to_qiskit(machine, QSimulationOptions(skip_qutip=True)) + assert "context-update actions are annotations only" not in script + + def test_mermaid_renders_transition_label_unchanged(self): + """Mermaid uses the action name on the transition arrow as-is.""" + from q_orca.compiler.mermaid import compile_to_mermaid + + effect = "if bits[0] == 1: theta[0] -= eta" + source = _base_machine_with_update(effect) + machine = _machine(source) + mermaid = compile_to_mermaid(machine) + # The action label is "gradient_step" — the raw effect string + # should NOT leak into the mermaid output. + assert "gradient_step" in mermaid + assert "theta[0]" not in mermaid