feat(OMN-20680): add linear-comments and hourly-tick decision compute nodes

contracts/OMN-20680.yaml

@@ -0,0 +1,51 @@
+---
+schema_version: "1.0.0"
+ticket_id: "OMN-20680"
+title: "The linear-comments sweep and hourly tick decisions run as omnimarket compute nodes"
+summary: "node_linear_comments_decision_compute ports the pure block of the linear-comments-hourly workflow script and node_hourly_tick_decision_compute holds the tick's probe and record decisions. Both are contract-declared COMPUTE nodes dispatched over the bus, with parity against captured outputs of the old code. Repointing callers and deleting the old script are separate changes."
+is_seam_ticket: false
+interface_change: false
+interfaces_touched: []
+requirements:
+  - id: "req-linear-comments-decision-node"
+    statement: |-
+      The linear-comments sweep's deterministic decisions (author classification, draft-or-not, prompt, citation check, truncation, held classification, held tally, degraded flag, metrics row) run in node_linear_comments_decision_compute and match the old JS pure block.
+    acceptance:
+      - id: "AC1"
+        statement: |-
+          AC1: Every decision kind returns the old block's captured output for the same input -- falsifier: tests/nodes/node_linear_comments_decision_compute/test_parity_linear_comments_decision_compute.py against tests/fixtures/linear_comments_hourly_parity.json, which fails on any differing decision
+      - id: "AC2"
+        statement: |-
+          AC2: The node is dispatched through its contract on the RuntimeLocal in-memory bus and an invalid request is refused with no result -- falsifier: tests/nodes/node_linear_comments_decision_compute/test_golden_chain_linear_comments_decision_compute.py
+  - id: "req-hourly-tick-decision-node"
+    statement: |-
+      The hourly tick's probe and record decisions run in node_hourly_tick_decision_compute, with no clock and no file reads, and match the pre-conversion tick.
+    acceptance:
+      - id: "AC3"
+        statement: |-
+          AC3: probe then record over the bus returns the pre-conversion tick's window state, checkpoint line, sweep request, completion line, phase, degraded class and fire-result payload for every captured case -- falsifier: tests/nodes/node_hourly_tick_decision_compute/test_golden_chain_hourly_tick_decision_compute.py against tests/fixtures/hourly_tick_parity.json
+evidence_requirements:
+  - kind: "tests"
+    description: "Parity, golden-chain and error-chain tests for both decision compute nodes."
+    command: "uv run pytest tests/nodes/node_linear_comments_decision_compute tests/nodes/node_hourly_tick_decision_compute -q -p no:randomly"
+emergency_bypass:
+  enabled: false
+  justification: ""
+  follow_up_ticket_id: ""
+dod_evidence:
+  - id: "dod-omn20680-linear-comments-decision"
+    description: "node_linear_comments_decision_compute matches the old JS pure block and runs over the bus."
+    source: "manual"
+    checks:
+      - check_type: "test_passes"
+        check_value: "uv run pytest tests/nodes/node_linear_comments_decision_compute/test_parity_linear_comments_decision_compute.py tests/nodes/node_linear_comments_decision_compute/test_golden_chain_linear_comments_decision_compute.py -q -p no:randomly"
+        cwd: "${OMNI_HOME}/omnimarket"
+    binds_ac: ["AC1", "AC2"]
+  - id: "dod-omn20680-hourly-tick-decision"
+    description: "node_hourly_tick_decision_compute matches the pre-conversion tick and runs probe then record over the bus."
+    source: "manual"
+    checks:
+      - check_type: "test_passes"
+        check_value: "uv run pytest tests/nodes/node_hourly_tick_decision_compute/test_golden_chain_hourly_tick_decision_compute.py -q -p no:randomly"
+        cwd: "${OMNI_HOME}/omnimarket"
+    binds_ac: ["AC3"]

pyproject.toml

@@ -271,6 +271,8 @@ node_lab_fill_effect = "omnimarket.nodes.node_lab_fill_effect"
 node_delegation_split_compute = "omnimarket.nodes.node_delegation_split_compute"
 node_delegation_availability_report_compute = "omnimarket.nodes.node_delegation_availability_report_compute"
 node_delegation_size_band_compute = "omnimarket.nodes.node_delegation_size_band_compute"
+node_linear_comments_decision_compute = "omnimarket.nodes.node_linear_comments_decision_compute"
+node_hourly_tick_decision_compute = "omnimarket.nodes.node_hourly_tick_decision_compute"
 node_delegation_recombine_compute = "omnimarket.nodes.node_delegation_recombine_compute"
 node_projection_worktree_reconcile = "omnimarket.nodes.node_projection_worktree_reconcile"
 node_worktree_reconcile_compute = "omnimarket.nodes.node_worktree_reconcile_compute"

src/omnimarket/nodes/node_hourly_tick_decision_compute/__init__.py

@@ -0,0 +1,14 @@
+# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Hourly tick decision compute node (OMN-20680)."""
+
+from omnimarket.nodes.node_hourly_tick_decision_compute.handlers.handler_hourly_tick_decision import (
+    HandlerHourlyTickDecision,
+)
+
+
+class NodeHourlyTickDecisionCompute(HandlerHourlyTickDecision):
+    """ONEX entrypoint for the tick's deterministic decisions."""
+
+
+__all__ = ["HandlerHourlyTickDecision", "NodeHourlyTickDecisionCompute"]

src/omnimarket/nodes/node_hourly_tick_decision_compute/contract.yaml

@@ -0,0 +1,81 @@
+# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+name: node_hourly_tick_decision_compute
+node_type: compute
+runtime_profiles: [main]
+contract_version: {major: 1, minor: 0, patch: 0}
+node_version: "1.0.0"
+description: >-
+  Definition-B deterministic decisions of the hourly tick (OMN-20680). probe turns the
+  checkpoint file, the sweep pointer and the ledger contents (read by the caller) into the
+  window state, the checkpoint line and the sweep request; record turns the state and the
+  sweep result into the completion line, the phase, the degraded class and the fire-result
+  payload. Reads no files, has no clock (the caller supplies now), mutates nothing. Every
+  refusal the retired tick made before a write is a refusal here, so a caller that gets one
+  has written nothing. Behaviour-preserving port of the retired tick, pinned by 60 captured
+  cases.
+input_model:
+  name: ModelHourlyTickDecisionRequest
+  module: omnimarket.nodes.node_hourly_tick_decision_compute.models.model_hourly_tick_decision
+output_model:
+  name: ModelHourlyTickDecisionResult
+  module: omnimarket.nodes.node_hourly_tick_decision_compute.models.model_hourly_tick_decision
+handler:
+  module: omnimarket.nodes.node_hourly_tick_decision_compute.handlers.handler_hourly_tick_decision
+  class: HandlerHourlyTickDecision
+  input_model: omnimarket.nodes.node_hourly_tick_decision_compute.models.model_hourly_tick_decision.ModelHourlyTickDecisionRequest
+  output_model: omnimarket.nodes.node_hourly_tick_decision_compute.models.model_hourly_tick_decision.ModelHourlyTickDecisionResult
+handler_routing:
+  routing_strategy: operation_match
+  default_handler: omnimarket.nodes.node_hourly_tick_decision_compute.handlers.handler_hourly_tick_decision:HandlerHourlyTickDecision
+  handlers:
+    - operation: decide_hourly_tick
+      handler:
+        name: HandlerHourlyTickDecision
+        module: omnimarket.nodes.node_hourly_tick_decision_compute.handlers.handler_hourly_tick_decision
+descriptor:
+  node_archetype: compute
+  purity: pure
+  idempotent: true
+  timeout_ms: 1000
+inputs:
+  kind: {type: str, required: true, enum: [probe, record]}
+  now: {type: str, required: false, description: ISO-Z time of the tick, for probe.}
+  checkpoint_text: {type: str, required: false, description: Contents of the checkpoint file, for probe.}
+  sweep_pointer_text: {type: str, required: false, description: Contents of the sweep pointer file, for probe.}
+  ledger_text: {type: str, required: false, description: Contents of the rolling ledger, for probe.}
+  tailLines: {type: int, required: false}
+  facts: {type: str, required: false, description: Caller facts; take precedence over the derived facts.}
+  taskType: {type: str, required: false}
+  skipSweep: {type: bool, required: false}
+  state: {type: object, required: false, description: The state probe returned, for record.}
+  sweep: {type: object, required: false, description: The sweep result, for record.}
+  runId: {type: str, required: false}
+  fireId: {type: str, required: false}
+outputs:
+  kind: {type: str}
+  state: {type: object}
+  checkpoint_line: {type: str}
+  facts_source: {type: str}
+  sweep_request: {type: object}
+  completion_line: {type: str}
+  phase: {type: str, enum: [prepared, completed, degraded]}
+  degraded: {type: bool}
+  class: {type: str}
+  result_payload: {type: object}
+terminal_event: onex.evt.omnimarket.hourly-tick-decision-completed.v1
+runtime_dispatch:
+  command_topic: onex.cmd.omnimarket.hourly-tick-decision-requested.v1
+event_bus:
+  subscribe_topics:
+    - onex.cmd.omnimarket.hourly-tick-decision-requested.v1
+  publish_topics:
+    - onex.evt.omnimarket.hourly-tick-decision-completed.v1
+externally_consumed_topics:
+  - onex.evt.omnimarket.hourly-tick-decision-completed.v1
+metadata:
+  transport_type: inmemory
+  description: Hourly tick decisions
+  author: OmniNode Team
+  created: "2026-10-08"
+  related_tickets: [OMN-20680]

src/omnimarket/nodes/node_hourly_tick_decision_compute/handlers/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT

src/omnimarket/nodes/node_hourly_tick_decision_compute/handlers/handler_hourly_tick_decision.py

@@ -0,0 +1,215 @@
+# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Definition-B decisions of the hourly tick (OMN-20680).
+
+The retired tick mixed these decisions with host file reads and writes. Here the caller
+reads the checkpoint file, the sweep pointer and the ledger and passes their contents; the
+handler returns what to write. Formats are byte-identical to the retired tick (60 captured
+cases in tests/fixtures/hourly_tick_parity.json).
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Literal
+
+from omnibase_core.types import JsonType
+
+from omnimarket.nodes.node_hourly_tick_decision_compute.models.model_hourly_tick_decision import (
+    EnumHourlyTickDecisionKind,
+    ModelHourlyTickDecisionRequest,
+    ModelHourlyTickDecisionResult,
+    ModelSweepRequest,
+    ModelSweepResult,
+    ModelTickState,
+)
+
+HELD_CLASS_ORDER = ("floor_refused", "truncated", "unsourced", "transport", "timeout")
+TS = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
+
+
+def checkpoint_line(p: ModelTickState) -> str:
+    return (
+        f"- {p.now} 60m checkpoint: ledger rows since {p.prev}: {p.rows}; "
+        f"TERMINAL {p.terminal} [{p.lanes}]; sweep since={p.sweep_since}"
+    )
+
+
+def tick_outcome(sweep: ModelSweepResult | None) -> dict[str, str | None]:
+    """Degraded when something needed a reply and nothing was accepted; ties follow HELD_CLASS_ORDER."""
+    if sweep is None or not (sweep.needing and sweep.accepted == 0):
+        return {"phase": "completed", "class": None}
+    counts = sweep.held_by_class or {}
+    best = max(HELD_CLASS_ORDER, key=lambda c: counts.get(c, 0))
+    return {
+        "phase": "degraded",
+        "class": best if counts.get(best, 0) > 0 else "unclassified",
+    }
+
+
+def completion_line(
+    p: ModelTickState, sweep: ModelSweepResult | None, run_id: str
+) -> str:
+    outcome = tick_outcome(sweep)
+    counts = ", ".join(
+        f"{field} {value if value is not None else 'n/a'}"
+        for field in ("collected", "needing", "accepted", "held")
+        for value in [getattr(sweep, field) if sweep is not None else None]
+    )
+    phase = f"phase={outcome['phase']}" + (
+        f" class={outcome['class']}" if outcome["class"] else ""
+    )
+    drafts = sweep.linear_file if sweep is not None and sweep.linear_file else "none"
+    return (
+        f"- {p.now} hourly-tick run={run_id or 'unrecorded'}: checkpoint rows {p.rows} / "
+        f"TERMINAL {p.terminal}; sweep since={p.sweep_since} {counts}, {phase}; drafts {drafts}; "
+        "NOT POSTED (per-message operator approval required)"
+    )
+
+
+def result_payload(fire_id: str, sweep: ModelSweepResult | None) -> dict[str, JsonType]:
+    return {
+        "fire_id": fire_id,
+        "workflow": "hourly-tick",
+        **tick_outcome(sweep),
+        **{
+            field: getattr(sweep, field) if sweep is not None else None
+            for field in ("needing", "accepted", "held", "held_by_class")
+        },
+    }
+
+
+def facts_from_tail(tail: list[str], limit: int) -> str:
+    lines = [line for line in tail if line.strip()][-limit:]
+    if not lines:
+        raise ValueError("hourly-tick: no facts supplied and checkpoint tail is empty")
+    return "\n".join(
+        [
+            "Ground-truth facts, derived from the most recent orchestration "
+            "checkpoint lines, oldest first.",
+            "Each line is a timestamped statement of what a lane reported "
+            "or what the operator ruled.",
+            "Cite only what appears below. Anything not here is not a fact you may state.",
+            "",
+            *[line if line.startswith("- ") else f"- {line}" for line in lines],
+        ]
+    )
+
+
+def probe_window(
+    *,
+    now: str,
+    checkpoint_text: str,
+    sweep_pointer_text: str,
+    ledger_text: str,
+    tail_lines: int,
+) -> ModelTickState:
+    """Window after the previous checkpoint through now. The ledger row count is the positive control."""
+    tail = checkpoint_text.splitlines()
+    checkpoints = [
+        m.group(1)
+        for line in tail
+        if (m := re.match(r"^- ([0-9T:Z-]*) 60m checkpoint:", line))
+    ]
+    if not checkpoints:
+        raise ValueError("hourly-tick: probe field PREV is not an ISO-Z timestamp")
+    prev = checkpoints[-1]
+    since = "".join(sweep_pointer_text.split())
+    if not TS.fullmatch(since):
+        raise ValueError(
+            "hourly-tick: probe field SWEEP_SINCE is not an ISO-Z timestamp"
+        )
+    ledger = ledger_text.splitlines()
+    timestamped = [
+        line for line in ledger if len(line) >= 20 and TS.fullmatch(line[:20])
+    ]
+    rows = [line for line in timestamped if prev < line[:20] <= now]
+    terminal = [line for line in rows if "| TERMINAL |" in line]
+    lanes: set[str] = set()
+    for line in terminal:
+        for field in line.split("|"):
+            if match := re.match(r"^lane=[A-Za-z0-9_.:-]+", field.strip()):
+                lanes.add(match.group())
+                break
+    ledger_total = sum(bool(re.match(r"^\d{4}-\d{2}-\d{2}T", line)) for line in ledger)
+    if ledger_total <= 0:
+        raise ValueError(
+            "hourly-tick: ledger positive control failed: no timestamped rows"
+        )
+    return ModelTickState(
+        now=now,
+        prev=prev,
+        sweep_since=since,
+        rows=len(rows),
+        terminal=len(terminal),
+        lanes=",".join(sorted(lanes)),
+        ledger_total=ledger_total,
+        tail=[line.strip() for line in tail[-tail_lines:] if line.strip()],
+    )
+
+
+class HandlerHourlyTickDecision:
+    """Stateless compute: probe the window, or record the sweep outcome."""
+
+    def handle(
+        self, request: ModelHourlyTickDecisionRequest
+    ) -> ModelHourlyTickDecisionResult:
+        if request.kind is EnumHourlyTickDecisionKind.PROBE:
+            assert request.now is not None
+            assert request.checkpoint_text is not None
+            assert request.sweep_pointer_text is not None
+            assert request.ledger_text is not None
+            state = probe_window(
+                now=request.now,
+                checkpoint_text=request.checkpoint_text,
+                sweep_pointer_text=request.sweep_pointer_text,
+                ledger_text=request.ledger_text,
+                tail_lines=request.tail_lines,
+            )
+            facts = request.facts or facts_from_tail(state.tail, request.tail_lines)
+            facts_source = (
+                "caller-supplied"
+                if request.facts
+                else f"derived from the last {request.tail_lines} checkpoint lines"
+            )
+        else:
+            assert request.state is not None
+            state = request.state
+            facts = request.facts
+            facts_source = None
+        skipped = (
+            request.skip_sweep and request.kind is EnumHourlyTickDecisionKind.PROBE
+        )
+        probing = request.kind is EnumHourlyTickDecisionKind.PROBE
+        sweep_request = (
+            ModelSweepRequest(
+                since=state.sweep_since, facts=facts, task_type=request.task_type
+            )
+            if probing and not skipped
+            else None
+        )
+        outcome = tick_outcome(request.sweep)
+        completing = not probing or skipped
+        phase: Literal["prepared", "completed", "degraded"] = "prepared"
+        if completing:
+            phase = "degraded" if outcome["phase"] == "degraded" else "completed"
+        return ModelHourlyTickDecisionResult(
+            kind=request.kind,
+            state=state,
+            checkpoint_line=checkpoint_line(state),
+            facts_source=facts_source,
+            sweep_request=sweep_request,
+            completion_line=(
+                completion_line(state, request.sweep, request.run_id or request.fire_id)
+                if completing
+                else None
+            ),
+            phase=phase,
+            degraded=phase == "degraded",
+            held_class=outcome["class"] if completing else None,
+            result_payload=(
+                result_payload(request.fire_id, request.sweep)
+                if completing and request.fire_id
+                else None
+            ),
+        )