diff --git a/db/schema.sql b/db/schema.sql
index 3c21596..67fdafe 100644
--- a/db/schema.sql
+++ b/db/schema.sql
@@ -12,6 +12,23 @@ CREATE TABLE IF NOT EXISTS material_consumption_snapshots (
 CREATE INDEX IF NOT EXISTS material_consumption_snapshots_machine_time_idx
     ON material_consumption_snapshots (calculation_id, machine_id, timestamp);
 
+-- Channel identities are intentionally independent of material assignment.
+CREATE TABLE IF NOT EXISTS channel_material_consumption_snapshots (
+    timestamp timestamptz NOT NULL,
+    calculation_id text NOT NULL,
+    machine_id text NOT NULL,
+    extruder text NOT NULL,
+    doser_channel text NOT NULL,
+    production_order text NOT NULL,
+    run_id text NOT NULL,
+    cumulative_consumption_kg double precision NOT NULL,
+    material_number text NULL,
+    material_name text NULL,
+    material_mapping_status text NOT NULL CHECK (material_mapping_status IN ('ASSIGNED', 'UNAVAILABLE', 'AMBIGUOUS', 'UNMAPPED')),
+    percentage_sum_valid boolean NOT NULL,
+    PRIMARY KEY (calculation_id, machine_id, extruder, doser_channel, production_order, timestamp)
+);
+
 CREATE TABLE IF NOT EXISTS material_efficiency_snapshots (
     erp_feedback_timestamp timestamptz NOT NULL,
     material_snapshot_timestamp timestamptz NOT NULL,
@@ -70,3 +87,42 @@ CREATE TABLE IF NOT EXISTS power_meter_monthly_reports (
     generated_at timestamptz NOT NULL DEFAULT now(),
     PRIMARY KEY (meter, month)
 );
+
+-- Mutable ENLYZE downtime source records. external_id is the stable ENLYZE UUID.
+CREATE TABLE IF NOT EXISTS production_downtime_events (
+    external_id text PRIMARY KEY,
+    machine_id text NOT NULL,
+    production_order text NULL,
+    source_type text NOT NULL,
+    source_start timestamptz NOT NULL,
+    source_end timestamptz NULL,
+    attributed_start timestamptz NULL,
+    attributed_end timestamptz NULL,
+    attributed_duration_seconds double precision NULL,
+    reason_id text NULL,
+    reason_name text NULL,
+    reason_description text NULL,
+    reason_group text NULL,
+    category text NOT NULL CHECK (category IN ('PLANNED', 'UNPLANNED', 'UNKNOWN')),
+    comment text NULL,
+    source_updated_at timestamptz NULL,
+    last_reconciled_at timestamptz NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS production_downtime_events_machine_attributed_time_idx
+    ON production_downtime_events (machine_id, attributed_start);
+CREATE INDEX IF NOT EXISTS production_downtime_events_production_order_idx
+    ON production_downtime_events (production_order);
+CREATE INDEX IF NOT EXISTS production_downtime_events_category_idx
+    ON production_downtime_events (category);
+CREATE INDEX IF NOT EXISTS production_downtime_events_unresolved_idx
+    ON production_downtime_events (machine_id, source_start)
+    WHERE source_end IS NULL OR category = 'UNKNOWN';
+
+-- One observed ERP order boundary per machine, used to clip open ENLYZE downtimes.
+CREATE TABLE IF NOT EXISTS production_order_attribution_state (
+    machine_id text PRIMARY KEY,
+    production_order text NOT NULL,
+    started_at timestamptz NOT NULL,
+    ended_at timestamptz NULL
+);
diff --git a/src/production_analytics/cli/__main__.py b/src/production_analytics/cli/__main__.py
index 367ff1a..de828b0 100644
--- a/src/production_analytics/cli/__main__.py
+++ b/src/production_analytics/cli/__main__.py
@@ -128,10 +128,25 @@ def _parser() -> argparse.ArgumentParser:
     monthly.add_argument("--meter")
     monthly.add_argument("--month", required=True, help="YYYY-MM")
     monthly.add_argument("--db-secrets-file", type=Path, default=Path("secrets/postgres.env"))
+
+    channel_material = runners.add_parser(
+        "channel-material-poll", help="Continuously poll channel-level material consumption"
+    )
+    channel_material.add_argument("--config", type=Path, required=True)
+    channel_material.add_argument("--poll-interval-seconds", type=float, default=10.0)
+    channel_material.add_argument(
+        "--state-directory", type=Path, default=Path("data/state/channel-material")
+    )
+    channel_material.add_argument("--secrets-file", type=Path, default=Path("secrets/enlyze.env"))
+    channel_material.add_argument("--once", action="store_true", help="Run one poll cycle and exit")
     efficiency = runners.add_parser("material-efficiency", help="Persist ERP feedback KPI points")
     efficiency.add_argument("--config", type=Path, required=True)
     efficiency.add_argument("--secrets-file", type=Path, default=Path("secrets/enlyze.env"))
     efficiency.add_argument("--erp-secrets-file", type=Path, default=Path("secrets/erp.env"))
+    downtime = runners.add_parser("downtime-reconcile", help="Reconcile ENLYZE downtime events")
+    downtime.add_argument("--config", type=Path, required=True)
+    downtime.add_argument("--secrets-file", type=Path, default=Path("secrets/enlyze.env"))
+    downtime.add_argument("--erp-secrets-file", type=Path, default=Path("secrets/erp.env"))
     return parser
 
 
@@ -174,6 +189,35 @@ def _run_material(args: argparse.Namespace) -> int:
     return 0
 
 
+def _run_channel_material(args: argparse.Namespace) -> int:
+    from production_analytics.calculations.channel_config import load_channel_material_calculation
+    from production_analytics.calculations.config import CalculationConfigError
+    from production_analytics.enlyze.exploration import ConfigurationError
+    from production_analytics.service.channel_material_runtime import build_channel_material_runner
+
+    try:
+        calculation = load_channel_material_calculation(args.config)
+        runner = build_channel_material_runner(
+            calculation,
+            poll_interval_seconds=args.poll_interval_seconds,
+            state_directory=args.state_directory,
+            secrets_file=args.secrets_file,
+        )
+        if args.once:
+            runner.run_once()
+        else:
+            runner.run()
+    except (CalculationConfigError, ConfigurationError) as exc:
+        print(f"Config/startup error: {exc}", file=sys.stderr)
+        return 2
+    except KeyboardInterrupt:
+        return 0
+    except Exception as exc:
+        print(f"Startup/runtime error: {type(exc).__name__}", file=sys.stderr)
+        return 1
+    return 0
+
+
 def _run_material_efficiency(args: argparse.Namespace) -> int:
     from production_analytics.calculations.config import CalculationConfigError
     from production_analytics.enlyze.exploration import ConfigurationError
@@ -263,13 +307,32 @@ def _run_monthly_report(args: argparse.Namespace) -> int:
         return 2
 
 
+def _run_downtime(args: argparse.Namespace) -> int:
+    from production_analytics.service.downtime_runtime import build_downtime_runner
+
+    try:
+        build_downtime_runner(
+            args.config,
+            secrets_file=args.secrets_file,
+            erp_secrets_file=args.erp_secrets_file,
+        ).run()
+    except Exception as exc:
+        print(f"Config/startup/runtime error: {type(exc).__name__}", file=sys.stderr)
+        return 1
+    return 0
+
+
 def main(argv: Sequence[str] | None = None) -> int:
     args = _parser().parse_args(argv)
     if args.namespace == "run":
+        if args.command == "channel-material-poll":
+            return _run_channel_material(args)
         if args.command == "material-efficiency":
             return _run_material_efficiency(args)
         if args.command == "power-meter":
             return _run_power_meter(args)
+        if args.command == "downtime-reconcile":
+            return _run_downtime(args)
         return _run_material(args)
     if args.namespace == "report":
         return _run_monthly_report(args)