feat: add repair parts tracking

backend/alembic/versions/0014_repair_parts_tracking.py

@@ -0,0 +1,590 @@
+"""add repair parts tracking
+
+Revision ID: 0014_repair_parts_tracking
+Revises: 0013_control_board_replacement
+Create Date: 2026-08-02
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+
+revision: str = "0014_repair_parts_tracking"
+down_revision: Union[str, None] = "0013_control_board_replacement"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+COMPONENT_TYPES = ("control_board", "hashboard", "power_supply", "fan")
+CONSUMABLE_UNITS = ("each", "gram", "millilitre", "metre", "centimetre")
+REPAIR_REASONS = (
+    "HARDWARE_FAILURE", "LOW_HASHRATE", "CONTROL_BOARD", "HASHBOARD",
+    "PSU", "FAN", "OVERHEATING", "NETWORKING",
+    "PREVENTIVE_MAINTENANCE", "INSPECTION", "OTHER",
+)
+REPLACEMENT_REASONS = (
+    "failed", "upgrade", "diagnostic_swap", "preventive_maintenance",
+    "temporary_donor", "unknown",
+)
+DISPOSITIONS = (
+    "repair_inventory", "sent_for_repair", "scrapped",
+    "retained_for_diagnosis", "returned_to_vendor", "unknown",
+)
+
+
+def _quoted(values: Sequence[str]) -> str:
+    return ", ".join(f"'{value}'" for value in values)
+
+
+def _create_event_constraints(include_parts: bool) -> None:
+    event_types = [
+        "REPAIR_STARTED",
+        "INITIAL_NOTE",
+        "REPAIR_COMPLETED",
+        "CONTROL_BOARD_REPLACED",
+    ]
+    if include_parts:
+        event_types.extend(("REPAIR_PART_RECORDED", "REPAIR_PART_VOIDED"))
+
+    op.create_check_constraint(
+        "ck_repair_events_type",
+        "repair_events",
+        f"event_type IN ({_quoted(event_types)})",
+    )
+
+    parts_clause = ""
+    if include_parts:
+        parts_clause = (
+            ") OR (event_type = 'REPAIR_PART_RECORDED' "
+            "AND payload ?& ARRAY['repair_part_id', 'entry_kind', 'notes'] "
+            "AND payload ->> 'repair_part_id' ~ "
+            "'^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-"
+            "[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$' "
+            "AND (jsonb_typeof(payload -> 'notes') = 'null' OR ("
+            "jsonb_typeof(payload -> 'notes') = 'string' "
+            "AND trim(payload ->> 'notes') <> '' "
+            "AND char_length(payload ->> 'notes') <= 4000)) "
+            "AND ((payload ->> 'entry_kind' = 'consumable' "
+            "AND payload ?& ARRAY['description', 'quantity', 'unit'] "
+            "AND NOT payload ? 'component_id' "
+            "AND jsonb_typeof(payload -> 'description') = 'string' "
+            "AND trim(payload ->> 'description') <> '' "
+            "AND char_length(payload ->> 'description') <= 255 "
+            "AND jsonb_typeof(payload -> 'quantity') = 'string' "
+            "AND payload ->> 'quantity' ~ '^[0-9]{1,9}(\\.[0-9]{1,3})?$' "
+            "AND (payload ->> 'quantity')::numeric > 0 "
+            f"AND payload ->> 'unit' IN ({_quoted(CONSUMABLE_UNITS)}) "
+            "AND (payload ->> 'unit' <> 'each' OR "
+            "(payload ->> 'quantity')::numeric = "
+            "trunc((payload ->> 'quantity')::numeric))) OR ("
+            "payload ->> 'entry_kind' = 'tracked_component' "
+            "AND payload ?& ARRAY['component_id', 'component_type', "
+            "'component_asset_id', 'component_manufacturer', "
+            "'component_model', 'component_manufacturer_serial'] "
+            "AND NOT payload ? 'description' "
+            "AND NOT payload ? 'quantity' "
+            "AND NOT payload ? 'unit' "
+            "AND payload ->> 'component_id' ~ "
+            "'^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-"
+            "[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$' "
+            f"AND payload ->> 'component_type' IN ({_quoted(COMPONENT_TYPES)}) "
+            "AND jsonb_typeof(payload -> 'component_asset_id') IN ('string', 'null') "
+            "AND jsonb_typeof(payload -> 'component_manufacturer') IN ('string', 'null') "
+            "AND jsonb_typeof(payload -> 'component_model') IN ('string', 'null') "
+            "AND jsonb_typeof(payload -> 'component_manufacturer_serial') "
+            "IN ('string', 'null'))))"
+            " OR (event_type = 'REPAIR_PART_VOIDED' "
+            "AND payload ?& ARRAY['repair_part_id', 'reason'] "
+            "AND payload ->> 'repair_part_id' ~ "
+            "'^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-"
+            "[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$' "
+            "AND jsonb_typeof(payload -> 'reason') = 'string' "
+            "AND trim(payload ->> 'reason') <> '' "
+            "AND char_length(payload ->> 'reason') <= 1000"
+        )
+
+    op.create_check_constraint(
+        "ck_repair_events_payload_shape",
+        "repair_events",
+        "(event_type = 'REPAIR_STARTED' "
+        "AND payload ?& ARRAY['reason', 'from_lifecycle', 'to_lifecycle'] "
+        f"AND payload ->> 'reason' IN ({_quoted(REPAIR_REASONS)}) "
+        "AND payload ->> 'from_lifecycle' IN ('active', 'spare') "
+        "AND payload ->> 'to_lifecycle' = 'in_repair') OR ("
+        "event_type = 'INITIAL_NOTE' "
+        "AND jsonb_typeof(payload -> 'note') = 'string' "
+        "AND trim(payload ->> 'note') <> '') OR ("
+        "event_type = 'REPAIR_COMPLETED' "
+        "AND payload ?& ARRAY['previous_miner_lifecycle', 'resulting_miner_lifecycle'] "
+        "AND payload ->> 'previous_miner_lifecycle' = 'in_repair' "
+        "AND payload ->> 'resulting_miner_lifecycle' IN ('active', 'spare') "
+        "AND (NOT payload ? 'pre_repair_lifecycle' "
+        "OR payload ->> 'pre_repair_lifecycle' IN ('active', 'spare')) "
+        "AND (NOT payload ? 'completion_summary' OR ("
+        "jsonb_typeof(payload -> 'completion_summary') = 'string' "
+        "AND trim(payload ->> 'completion_summary') <> '' "
+        "AND char_length(payload ->> 'completion_summary') <= 4000))) OR ("
+        "event_type = 'CONTROL_BOARD_REPLACED' "
+        "AND payload ?& ARRAY['old_mac_address', 'new_mac_address', "
+        "'old_control_board_serial', 'new_control_board_serial', "
+        "'replacement_reason', 'removed_board_disposition'] "
+        "AND jsonb_typeof(payload -> 'old_mac_address') IN ('string', 'null') "
+        "AND jsonb_typeof(payload -> 'new_mac_address') = 'string' "
+        "AND payload ->> 'new_mac_address' ~ '^([0-9A-F]{2}:){5}[0-9A-F]{2}$' "
+        "AND regexp_replace(lower(payload ->> 'new_mac_address'), '[^0-9a-f]', '', 'g') "
+        "NOT IN ('000000000000', 'ffffffffffff') "
+        "AND get_byte(decode(regexp_replace(lower(payload ->> 'new_mac_address'), "
+        "'[^0-9a-f]', '', 'g'), 'hex'), 0) % 2 = 0 "
+        "AND (jsonb_typeof(payload -> 'old_control_board_serial') = 'null' OR ("
+        "jsonb_typeof(payload -> 'old_control_board_serial') = 'string' "
+        "AND trim(payload ->> 'old_control_board_serial') <> '' "
+        "AND char_length(payload ->> 'old_control_board_serial') <= 255)) "
+        "AND (jsonb_typeof(payload -> 'new_control_board_serial') = 'null' OR ("
+        "jsonb_typeof(payload -> 'new_control_board_serial') = 'string' "
+        "AND trim(payload ->> 'new_control_board_serial') <> '' "
+        "AND char_length(payload ->> 'new_control_board_serial') <= 255)) "
+        f"AND payload ->> 'replacement_reason' IN ({_quoted(REPLACEMENT_REASONS)}) "
+        f"AND payload ->> 'removed_board_disposition' IN ({_quoted(DISPOSITIONS)}) "
+        "AND (NOT payload ? 'notes' OR jsonb_typeof(payload -> 'notes') = 'null' OR ("
+        "jsonb_typeof(payload -> 'notes') = 'string' "
+        "AND trim(payload ->> 'notes') <> '' "
+        "AND char_length(payload ->> 'notes') <= 4000))"
+        f"{parts_clause})",
+    )
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+
+    connection.exec_driver_sql(
+        """
+        INSERT INTO permissions (key, description)
+        VALUES
+          ('components.read', 'View tracked component identity'),
+          ('components.create', 'Register tracked component identity'),
+          ('repairs.parts.manage', 'Record and void repair-part usage')
+        ON CONFLICT (key) DO UPDATE SET description = EXCLUDED.description;
+
+        INSERT INTO role_permissions (role_id, permission_id)
+        SELECT roles.id, permissions.id
+        FROM roles
+        JOIN permissions ON permissions.key IN (
+          'components.read', 'components.create', 'repairs.parts.manage'
+        )
+        WHERE roles.name IN ('Admin', 'Manager', 'Technician')
+        ON CONFLICT DO NOTHING;
+
+        INSERT INTO role_permissions (role_id, permission_id)
+        SELECT roles.id, permissions.id
+        FROM roles
+        JOIN permissions ON permissions.key = 'components.read'
+        WHERE roles.name = 'Read Only'
+        ON CONFLICT DO NOTHING;
+        """
+    )
+
+    op.create_table(
+        "components",
+        sa.Column(
+            "id", postgresql.UUID(as_uuid=True), primary_key=True,
+            server_default=sa.text("gen_random_uuid()"),
+        ),
+        sa.Column("component_type", sa.Text(), nullable=False),
+        sa.Column("asset_id", sa.Text(), nullable=True),
+        sa.Column("manufacturer", sa.Text(), nullable=True),
+        sa.Column("model", sa.Text(), nullable=True),
+        sa.Column("manufacturer_serial", sa.Text(), nullable=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "created_by", postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="RESTRICT"), nullable=False,
+        ),
+        sa.CheckConstraint(
+            f"component_type IN ({_quoted(COMPONENT_TYPES)})",
+            name="ck_components_type",
+        ),
+        sa.CheckConstraint(
+            "asset_id IS NULL OR (trim(asset_id) <> '' "
+            "AND asset_id = trim(asset_id) AND char_length(asset_id) <= 255)",
+            name="ck_components_asset_id",
+        ),
+        sa.CheckConstraint(
+            "manufacturer IS NULL OR (trim(manufacturer) <> '' "
+            "AND manufacturer = trim(manufacturer) "
+            "AND char_length(manufacturer) <= 255)",
+            name="ck_components_manufacturer",
+        ),
+        sa.CheckConstraint(
+            "model IS NULL OR (trim(model) <> '' AND model = trim(model) "
+            "AND char_length(model) <= 255)",
+            name="ck_components_model",
+        ),
+        sa.CheckConstraint(
+            "manufacturer_serial IS NULL OR (trim(manufacturer_serial) <> '' "
+            "AND manufacturer_serial = trim(manufacturer_serial) "
+            "AND char_length(manufacturer_serial) <= 255)",
+            name="ck_components_manufacturer_serial",
+        ),
+    )
+    op.create_index("idx_components_created_by", "components", ["created_by"])
+    op.create_index("idx_components_type", "components", ["component_type"])
+    op.create_index(
+        "uq_components_normalized_asset_id",
+        "components",
+        [sa.text("lower(trim(asset_id))")],
+        unique=True,
+        postgresql_where=sa.text("asset_id IS NOT NULL"),
+    )
+
+    op.create_table(
+        "repair_parts",
+        sa.Column(
+            "id", postgresql.UUID(as_uuid=True), primary_key=True,
+            server_default=sa.text("gen_random_uuid()"),
+        ),
+        sa.Column(
+            "repair_id", postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("repairs.id", ondelete="RESTRICT"), nullable=False,
+        ),
+        sa.Column("consumable_description", sa.Text(), nullable=True),
+        sa.Column("quantity", sa.Numeric(precision=12, scale=3), nullable=True),
+        sa.Column("unit", sa.Text(), nullable=True),
+        sa.Column(
+            "component_id", postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("components.id", ondelete="RESTRICT"), nullable=True,
+        ),
+        sa.Column("component_type_snapshot", sa.Text(), nullable=True),
+        sa.Column("component_asset_id_snapshot", sa.Text(), nullable=True),
+        sa.Column("component_manufacturer_snapshot", sa.Text(), nullable=True),
+        sa.Column("component_model_snapshot", sa.Text(), nullable=True),
+        sa.Column("component_manufacturer_serial_snapshot", sa.Text(), nullable=True),
+        sa.Column("notes", sa.Text(), nullable=True),
+        sa.Column(
+            "recorded_by", postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="RESTRICT"), nullable=False,
+        ),
+        sa.Column(
+            "recorded_at", sa.DateTime(timezone=True), nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "recorded_event_id", postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("repair_events.id", ondelete="RESTRICT"),
+            nullable=False,
+        ),
+        sa.Column(
+            "voided_by", postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="RESTRICT"), nullable=True,
+        ),
+        sa.Column("voided_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("void_reason", sa.Text(), nullable=True),
+        sa.Column(
+            "void_event_id", postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("repair_events.id", ondelete="RESTRICT"),
+            nullable=True,
+        ),
+        sa.CheckConstraint(
+            "((consumable_description IS NOT NULL AND quantity IS NOT NULL "
+            "AND unit IS NOT NULL AND component_id IS NULL "
+            "AND component_type_snapshot IS NULL "
+            "AND component_asset_id_snapshot IS NULL "
+            "AND component_manufacturer_snapshot IS NULL "
+            "AND component_model_snapshot IS NULL "
+            "AND component_manufacturer_serial_snapshot IS NULL) OR "
+            "(consumable_description IS NULL AND quantity IS NULL "
+            "AND unit IS NULL AND component_id IS NOT NULL "
+            "AND component_type_snapshot IS NOT NULL))",
+            name="ck_repair_parts_entry_xor",
+        ),
+        sa.CheckConstraint(
+            "quantity IS NULL OR quantity > 0",
+            name="ck_repair_parts_quantity_positive",
+        ),
+        sa.CheckConstraint(
+            "unit IS NULL OR unit <> 'each' OR quantity = trunc(quantity)",
+            name="ck_repair_parts_each_whole",
+        ),
+        sa.CheckConstraint(
+            f"unit IS NULL OR unit IN ({_quoted(CONSUMABLE_UNITS)})",
+            name="ck_repair_parts_unit",
+        ),
+        sa.CheckConstraint(
+            f"component_type_snapshot IS NULL OR "
+            f"component_type_snapshot IN ({_quoted(COMPONENT_TYPES)})",
+            name="ck_repair_parts_component_type",
+        ),
+        sa.CheckConstraint(
+            "consumable_description IS NULL OR ("
+            "consumable_description = trim(consumable_description) "
+            "AND trim(consumable_description) <> '' "
+            "AND char_length(consumable_description) <= 255)",
+            name="ck_repair_parts_description",
+        ),
+        sa.CheckConstraint(
+            "component_asset_id_snapshot IS NULL OR ("
+            "component_asset_id_snapshot = trim(component_asset_id_snapshot) "
+            "AND trim(component_asset_id_snapshot) <> '' "
+            "AND char_length(component_asset_id_snapshot) <= 255)",
+            name="ck_repair_parts_asset_snapshot",
+        ),
+        sa.CheckConstraint(
+            "component_manufacturer_snapshot IS NULL OR ("
+            "component_manufacturer_snapshot = trim(component_manufacturer_snapshot) "
+            "AND trim(component_manufacturer_snapshot) <> '' "
+            "AND char_length(component_manufacturer_snapshot) <= 255)",
+            name="ck_repair_parts_manufacturer_snapshot",
+        ),
+        sa.CheckConstraint(
+            "component_model_snapshot IS NULL OR ("
+            "component_model_snapshot = trim(component_model_snapshot) "
+            "AND trim(component_model_snapshot) <> '' "
+            "AND char_length(component_model_snapshot) <= 255)",
+            name="ck_repair_parts_model_snapshot",
+        ),
+        sa.CheckConstraint(
+            "component_manufacturer_serial_snapshot IS NULL OR ("
+            "component_manufacturer_serial_snapshot = "
+            "trim(component_manufacturer_serial_snapshot) "
+            "AND trim(component_manufacturer_serial_snapshot) <> '' "
+            "AND char_length(component_manufacturer_serial_snapshot) <= 255)",
+            name="ck_repair_parts_serial_snapshot",
+        ),
+        sa.CheckConstraint(
+            "notes IS NULL OR (notes = trim(notes) AND trim(notes) <> '' "
+            "AND char_length(notes) <= 4000)",
+            name="ck_repair_parts_notes",
+        ),
+        sa.CheckConstraint(
+            "(voided_by IS NULL AND voided_at IS NULL AND void_reason IS NULL "
+            "AND void_event_id IS NULL) OR "
+            "(voided_by IS NOT NULL AND voided_at IS NOT NULL "
+            "AND void_reason IS NOT NULL AND void_event_id IS NOT NULL)",
+            name="ck_repair_parts_void_state",
+        ),
+        sa.CheckConstraint(
+            "void_reason IS NULL OR (void_reason = trim(void_reason) "
+            "AND trim(void_reason) <> '' AND char_length(void_reason) <= 1000)",
+            name="ck_repair_parts_void_reason",
+        ),
+        sa.CheckConstraint(
+            "voided_at IS NULL OR voided_at >= recorded_at",
+            name="ck_repair_parts_void_chronology",
+        ),
+    )
+    op.create_index(
+        "idx_repair_parts_repair",
+        "repair_parts",
+        ["repair_id", "recorded_event_id", "id"],
+    )
+    op.create_index("idx_repair_parts_component", "repair_parts", ["component_id"])
+    op.create_index("idx_repair_parts_recorded_by", "repair_parts", ["recorded_by"])
+    op.create_index("idx_repair_parts_voided_by", "repair_parts", ["voided_by"])
+    op.create_index(
+        "uq_repair_parts_recorded_event",
+        "repair_parts",
+        ["recorded_event_id"],
+        unique=True,
+    )
+    op.create_index(
+        "uq_repair_parts_void_event",
+        "repair_parts",
+        ["void_event_id"],
+        unique=True,
+        postgresql_where=sa.text("void_event_id IS NOT NULL"),
+    )
+
+    op.drop_constraint("ck_repair_events_payload_shape", "repair_events", type_="check")
+    op.drop_constraint("ck_repair_events_type", "repair_events", type_="check")
+    _create_event_constraints(True)
+
+    connection.exec_driver_sql(
+        """
+        CREATE FUNCTION minerops_protect_component_identity()
+        RETURNS trigger LANGUAGE plpgsql AS $$
+        BEGIN
+          IF TG_OP = 'DELETE' THEN
+            RAISE EXCEPTION 'Durable component identity cannot be deleted';
+          END IF;
+          IF NEW.id <> OLD.id
+             OR NEW.component_type <> OLD.component_type
+             OR NEW.created_at <> OLD.created_at
+             OR NEW.created_by <> OLD.created_by THEN
+            RAISE EXCEPTION 'Component identity, type, and creation provenance are immutable';
+          END IF;
+          RETURN NEW;
+        END $$;
+        CREATE TRIGGER trg_protect_component_identity
+        BEFORE UPDATE OR DELETE ON components
+        FOR EACH ROW EXECUTE FUNCTION minerops_protect_component_identity();
+
+        CREATE FUNCTION minerops_protect_repair_event_history()
+        RETURNS trigger LANGUAGE plpgsql AS $$
+        BEGIN
+          RAISE EXCEPTION 'Repair event history is immutable';
+        END $$;
+        CREATE TRIGGER trg_protect_repair_event_history
+        BEFORE UPDATE OR DELETE ON repair_events
+        FOR EACH ROW EXECUTE FUNCTION minerops_protect_repair_event_history();
+
+        CREATE FUNCTION minerops_protect_repair_part_history()
+        RETURNS trigger LANGUAGE plpgsql AS $$
+        BEGIN
+          IF TG_OP = 'DELETE' THEN
+            RAISE EXCEPTION 'Repair-part history is immutable';
+          END IF;
+          IF NEW.id <> OLD.id
+             OR NEW.repair_id <> OLD.repair_id
+             OR NEW.consumable_description IS DISTINCT FROM OLD.consumable_description
+             OR NEW.quantity IS DISTINCT FROM OLD.quantity
+             OR NEW.unit IS DISTINCT FROM OLD.unit
+             OR NEW.component_id IS DISTINCT FROM OLD.component_id
+             OR NEW.component_type_snapshot IS DISTINCT FROM OLD.component_type_snapshot
+             OR NEW.component_asset_id_snapshot IS DISTINCT FROM OLD.component_asset_id_snapshot
+             OR NEW.component_manufacturer_snapshot IS DISTINCT FROM OLD.component_manufacturer_snapshot
+             OR NEW.component_model_snapshot IS DISTINCT FROM OLD.component_model_snapshot
+             OR NEW.component_manufacturer_serial_snapshot IS DISTINCT FROM OLD.component_manufacturer_serial_snapshot
+             OR NEW.notes IS DISTINCT FROM OLD.notes
+             OR NEW.recorded_by <> OLD.recorded_by
+             OR NEW.recorded_at <> OLD.recorded_at
+             OR NEW.recorded_event_id <> OLD.recorded_event_id THEN
+            RAISE EXCEPTION 'Original repair-part fields are immutable';
+          END IF;
+          IF OLD.voided_by IS NOT NULL OR OLD.voided_at IS NOT NULL
+             OR OLD.void_reason IS NOT NULL OR OLD.void_event_id IS NOT NULL THEN
+            RAISE EXCEPTION 'Voided repair-part history is immutable';
+          END IF;
+          IF NEW.voided_by IS NULL OR NEW.voided_at IS NULL
+             OR NEW.void_reason IS NULL OR NEW.void_event_id IS NULL THEN
+            RAISE EXCEPTION 'Repair-part void metadata must be written atomically';
+          END IF;
+          RETURN NEW;
+        END $$;
+        CREATE TRIGGER trg_protect_repair_part_history
+        BEFORE UPDATE OR DELETE ON repair_parts
+        FOR EACH ROW EXECUTE FUNCTION minerops_protect_repair_part_history();
+
+        CREATE FUNCTION minerops_validate_repair_part_history()
+        RETURNS trigger LANGUAGE plpgsql AS $$
+        DECLARE recorded repair_events%%ROWTYPE;
+        DECLARE voided repair_events%%ROWTYPE;
+        DECLARE component components%%ROWTYPE;
+        BEGIN
+          IF TG_OP = 'INSERT' AND (
+             NEW.voided_by IS NOT NULL OR NEW.voided_at IS NOT NULL
+             OR NEW.void_reason IS NOT NULL OR NEW.void_event_id IS NOT NULL) THEN
+            RAISE EXCEPTION 'A repair part must be recorded before it can be voided';
+          END IF;
+          SELECT * INTO recorded FROM repair_events WHERE id = NEW.recorded_event_id;
+          IF recorded.id IS NULL
+             OR recorded.repair_id <> NEW.repair_id
+             OR recorded.event_type <> 'REPAIR_PART_RECORDED'
+             OR recorded.created_by <> NEW.recorded_by
+             OR recorded.created_at <> NEW.recorded_at
+             OR recorded.payload ->> 'repair_part_id' <> NEW.id::text THEN
+            RAISE EXCEPTION 'Recorded repair-part event does not match its entry';
+          END IF;
+
+          IF TG_OP = 'INSERT' AND NEW.component_id IS NOT NULL THEN
+            SELECT * INTO component FROM components WHERE id = NEW.component_id;
+            IF component.id IS NULL
+               OR component.component_type <> NEW.component_type_snapshot
+               OR component.asset_id IS DISTINCT FROM NEW.component_asset_id_snapshot
+               OR component.manufacturer IS DISTINCT FROM NEW.component_manufacturer_snapshot
+               OR component.model IS DISTINCT FROM NEW.component_model_snapshot
+               OR component.manufacturer_serial IS DISTINCT FROM NEW.component_manufacturer_serial_snapshot THEN
+              RAISE EXCEPTION 'Tracked-component snapshot does not match the component';
+            END IF;
+          END IF;
+
+          IF NEW.void_event_id IS NOT NULL THEN
+            SELECT * INTO voided FROM repair_events WHERE id = NEW.void_event_id;
+            IF voided.id IS NULL
+               OR voided.repair_id <> NEW.repair_id
+               OR voided.event_type <> 'REPAIR_PART_VOIDED'
+               OR voided.created_by <> NEW.voided_by
+               OR voided.created_at <> NEW.voided_at
+               OR voided.payload ->> 'repair_part_id' <> NEW.id::text
+               OR voided.payload ->> 'reason' <> NEW.void_reason THEN
+              RAISE EXCEPTION 'Voided repair-part event does not match its entry';
+            END IF;
+          END IF;
+          RETURN NEW;
+        END $$;
+        CREATE TRIGGER trg_validate_repair_part_history
+        BEFORE INSERT OR UPDATE ON repair_parts
+        FOR EACH ROW EXECUTE FUNCTION minerops_validate_repair_part_history();
+        """
+    )
+
+
+def downgrade() -> None:
+    connection = op.get_bind()
+    connection.exec_driver_sql(
+        """
+        DO $$
+        DECLARE history_count bigint;
+        BEGIN
+          SELECT count(*) INTO history_count FROM components;
+          history_count := history_count + (SELECT count(*) FROM repair_parts);
+          history_count := history_count + (
+            SELECT count(*) FROM repair_events
+            WHERE event_type IN ('REPAIR_PART_RECORDED', 'REPAIR_PART_VOIDED')
+          );
+          IF history_count > 0 THEN
+            RAISE EXCEPTION USING
+              MESSAGE = format(
+                'Cannot downgrade Repair Parts Tracking: %s component, repair-part, or event records would be destroyed',
+                history_count
+              ),
+              HINT = 'Preserve or explicitly migrate Task 005 identity and history before downgrading; MinerOps will not delete or invent component, actor, timestamp, snapshot, or repair-part history.';
+          END IF;
+        END $$;
+        """
+    )
+
+    connection.exec_driver_sql(
+        """
+        DROP TRIGGER trg_validate_repair_part_history ON repair_parts;
+        DROP FUNCTION minerops_validate_repair_part_history();
+        DROP TRIGGER trg_protect_repair_part_history ON repair_parts;
+        DROP FUNCTION minerops_protect_repair_part_history();
+        DROP TRIGGER trg_protect_repair_event_history ON repair_events;
+        DROP FUNCTION minerops_protect_repair_event_history();
+        DROP TRIGGER trg_protect_component_identity ON components;
+        DROP FUNCTION minerops_protect_component_identity();
+        """
+    )
+
+    op.drop_constraint("ck_repair_events_payload_shape", "repair_events", type_="check")
+    op.drop_constraint("ck_repair_events_type", "repair_events", type_="check")
+    _create_event_constraints(False)
+
+    op.drop_index("uq_repair_parts_void_event", table_name="repair_parts")
+    op.drop_index("uq_repair_parts_recorded_event", table_name="repair_parts")
+    op.drop_index("idx_repair_parts_voided_by", table_name="repair_parts")
+    op.drop_index("idx_repair_parts_recorded_by", table_name="repair_parts")
+    op.drop_index("idx_repair_parts_component", table_name="repair_parts")
+    op.drop_index("idx_repair_parts_repair", table_name="repair_parts")
+    op.drop_table("repair_parts")
+
+    op.drop_index("uq_components_normalized_asset_id", table_name="components")
+    op.drop_index("idx_components_type", table_name="components")
+    op.drop_index("idx_components_created_by", table_name="components")
+    op.drop_table("components")
+
+    connection.exec_driver_sql(
+        """
+        DELETE FROM role_permissions
+        WHERE permission_id IN (
+          SELECT id FROM permissions
+          WHERE key IN ('components.read', 'components.create', 'repairs.parts.manage')
+        );
+        DELETE FROM permissions
+        WHERE key IN ('components.read', 'components.create', 'repairs.parts.manage');
+        """
+    )

backend/app/components/__init__.py

@@ -0,0 +1 @@
+"""Minimal durable component registry for repair-part references."""

backend/app/components/dependencies.py

@@ -0,0 +1,8 @@
+"""Dependency construction for the Components domain."""
+
+from app.components.repository import ComponentRepository
+from app.components.service import ComponentService
+
+
+def get_component_service() -> ComponentService:
+    return ComponentService(ComponentRepository())

backend/app/components/exceptions.py

@@ -0,0 +1,32 @@
+"""Domain exceptions for the minimal component registry."""
+
+
+class ComponentError(Exception):
+    """Base exception for component operations."""
+
+
+class ComponentNotFound(ComponentError):
+    """Raised when a component UUID does not exist."""
+
+    def __init__(self, component_id: object):
+        self.component_id = component_id
+        super().__init__(f"Component not found: {component_id}")
+
+
+class DuplicateComponentAssetID(ComponentError):
+    """Raised when a normalized component Asset ID already exists."""
+
+    def __init__(self, asset_id: str):
+        self.asset_id = asset_id
+        super().__init__(f"Component Asset ID already exists: {asset_id}")
+
+
+class ComponentValidationError(ComponentError):
+    """Raised when component metadata violates a domain rule."""
+
+
+class ComponentActorRequired(ComponentError):
+    """Raised when component creation lacks an authenticated actor."""
+
+    def __init__(self):
+        super().__init__("An authenticated actor is required to create a component")

backend/app/components/models.py

@@ -0,0 +1,36 @@
+"""Typed contracts for the minimal component registry."""
+
+from typing import Literal
+from uuid import UUID
+
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field
+
+
+ComponentType = Literal["control_board", "hashboard", "power_supply", "fan"]
+
+
+class ComponentCreateRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
+
+    component_type: ComponentType
+    asset_id: str | None = Field(default=None, max_length=255)
+    manufacturer: str | None = Field(default=None, max_length=255)
+    model: str | None = Field(default=None, max_length=255)
+    manufacturer_serial: str | None = Field(default=None, max_length=255)
+
+
+class ComponentDetail(BaseModel):
+    component_id: UUID
+    component_type: ComponentType
+    asset_id: str | None = None
+    manufacturer: str | None = None
+    model: str | None = None
+    manufacturer_serial: str | None = None
+    created_at: AwareDatetime
+    created_by: UUID
+
+
+class ComponentResponse(BaseModel):
+    data: ComponentDetail
+    meta: dict[str, object] = Field(default_factory=dict)
+    error: None = None

backend/app/components/repository.py

@@ -0,0 +1,143 @@
+"""PostgreSQL persistence for the minimal component registry."""
+
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+from uuid import UUID
+
+import psycopg
+from psycopg.errors import UniqueViolation
+from psycopg.rows import dict_row
+
+from app.components.exceptions import DuplicateComponentAssetID
+from app.config import settings
+
+
+ConnectionFactory = Callable[[], psycopg.Connection]
+COMPONENT_ASSET_ID_CONSTRAINT = "uq_components_normalized_asset_id"
+
+
+def default_connection_factory() -> psycopg.Connection:
+    return psycopg.connect(
+        host=settings.postgres_host,
+        port=settings.postgres_port,
+        dbname=settings.postgres_db,
+        user=settings.postgres_user,
+        password=settings.postgres_password,
+        connect_timeout=5,
+    )
+
+
+class ComponentRepository:
+    """Persistence boundary for minimal component identity."""
+
+    def __init__(
+        self,
+        connection_factory: ConnectionFactory | None = None,
+        *,
+        connection: psycopg.Connection | None = None,
+    ):
+        self._connection_factory = connection_factory or default_connection_factory
+        self._connection = connection
+
+    @contextmanager
+    def transaction(self) -> Iterator["ComponentRepository"]:
+        if self._connection is not None:
+            yield self
+            return
+
+        connection = self._connection_factory()
+        repository = ComponentRepository(
+            self._connection_factory,
+            connection=connection,
+        )
+        try:
+            yield repository
+            connection.commit()
+        except UniqueViolation as exc:
+            connection.rollback()
+            if exc.diag.constraint_name == COMPONENT_ASSET_ID_CONSTRAINT:
+                raise DuplicateComponentAssetID("normalized Asset ID") from exc
+            raise
+        except Exception:
+            connection.rollback()
+            raise
+        finally:
+            connection.close()
+
+    @contextmanager
+    def _connection_scope(self) -> Iterator[psycopg.Connection]:
+        if self._connection is not None:
+            yield self._connection
+            return
+
+        connection = self._connection_factory()
+        try:
+            yield connection
+        finally:
+            connection.close()
+
+    def create_component(
+        self,
+        *,
+        component_type: str,
+        asset_id: str | None,
+        manufacturer: str | None,
+        model: str | None,
+        manufacturer_serial: str | None,
+        created_by: UUID,
+    ) -> dict:
+        with self._connection_scope() as connection:
+            with connection.cursor(row_factory=dict_row) as cur:
+                cur.execute(
+                    """
+                    INSERT INTO components (
+                        component_type,
+                        asset_id,
+                        manufacturer,
+                        model,
+                        manufacturer_serial,
+                        created_by
+                    )
+                    VALUES (%s, %s, %s, %s, %s, %s)
+                    RETURNING
+                        id AS component_id,
+                        component_type,
+                        asset_id,
+                        manufacturer,
+                        model,
+                        manufacturer_serial,
+                        created_at,
+                        created_by
+                    """,
+                    (
+                        component_type,
+                        asset_id,
+                        manufacturer,
+                        model,
+                        manufacturer_serial,
+                        created_by,
+                    ),
+                )
+                return dict(cur.fetchone())
+
+    def get_component(self, component_id: UUID) -> dict | None:
+        with self._connection_scope() as connection:
+            with connection.cursor(row_factory=dict_row) as cur:
+                cur.execute(
+                    """
+                    SELECT
+                        id AS component_id,
+                        component_type,
+                        asset_id,
+                        manufacturer,
+                        model,
+                        manufacturer_serial,
+                        created_at,
+                        created_by
+                    FROM components
+                    WHERE id = %s
+                    """,
+                    (component_id,),
+                )
+                row = cur.fetchone()
+                return dict(row) if row else None

backend/app/components/router.py

@@ -0,0 +1,80 @@
+"""HTTP mapping for the minimal component registry."""
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from app.components.dependencies import get_component_service
+from app.components.exceptions import (
+    ComponentActorRequired,
+    ComponentNotFound,
+    ComponentValidationError,
+    DuplicateComponentAssetID,
+)
+from app.components.models import (
+    ComponentCreateRequest,
+    ComponentResponse,
+)
+from app.components.service import ComponentService
+from app.identity import require_permission
+
+
+router = APIRouter(prefix="/api/components", tags=["components"])
+require_component_read = require_permission("components.read")
+require_component_create = require_permission("components.create")
+
+ComponentServiceDependency = Annotated[
+    ComponentService,
+    Depends(get_component_service),
+]
+ComponentReadUser = Annotated[dict, Depends(require_component_read)]
+ComponentCreateUser = Annotated[dict, Depends(require_component_create)]
+
+
+@router.post(
+    "",
+    response_model=ComponentResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def create_component(
+    request: ComponentCreateRequest,
+    service: ComponentServiceDependency,
+    current_user: ComponentCreateUser,
+) -> ComponentResponse:
+    try:
+        component = service.create_component(
+            request,
+            actor_user_id=UUID(str(current_user["id"])),
+        )
+    except DuplicateComponentAssetID as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(exc),
+        ) from exc
+    except (ComponentActorRequired, ComponentValidationError) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(exc),
+        ) from exc
+    return ComponentResponse(data=component)
+
+
+@router.get(
+    "/{component_id}",
+    response_model=ComponentResponse,
+)
+def get_component(
+    component_id: UUID,
+    service: ComponentServiceDependency,
+    current_user: ComponentReadUser,
+) -> ComponentResponse:
+    del current_user
+    try:
+        component = service.get_component(component_id)
+    except ComponentNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Component not found",
+        ) from exc
+    return ComponentResponse(data=component)

backend/app/components/service.py

@@ -0,0 +1,75 @@
+"""Business rules for the minimal component registry."""
+
+from uuid import UUID
+
+from app.components.exceptions import (
+    ComponentActorRequired,
+    ComponentNotFound,
+    ComponentValidationError,
+    DuplicateComponentAssetID,
+)
+from app.components.models import ComponentCreateRequest, ComponentDetail
+from app.components.repository import ComponentRepository
+
+
+class ComponentService:
+    """Minimal component workflows independent of FastAPI."""
+
+    def __init__(self, repository: ComponentRepository):
+        self.repository = repository
+
+    @staticmethod
+    def _normalize_optional_text(value: str | None, field_name: str) -> str | None:
+        if value is None:
+            return None
+        normalized = str(value).strip()
+        if not normalized:
+            raise ComponentValidationError(
+                f"{field_name} must not be blank when supplied"
+            )
+        if len(normalized) > 255:
+            raise ComponentValidationError(
+                f"{field_name} must not exceed 255 characters"
+            )
+        return normalized
+
+    @staticmethod
+    def _map_component(component: dict) -> ComponentDetail:
+        return ComponentDetail(**component)
+
+    def create_component(
+        self,
+        request: ComponentCreateRequest,
+        *,
+        actor_user_id: UUID | None,
+    ) -> ComponentDetail:
+        if actor_user_id is None:
+            raise ComponentActorRequired()
+
+        asset_id = self._normalize_optional_text(request.asset_id, "asset_id")
+        values = {
+            "manufacturer": self._normalize_optional_text(
+                request.manufacturer, "manufacturer"
+            ),
+            "model": self._normalize_optional_text(request.model, "model"),
+            "manufacturer_serial": self._normalize_optional_text(
+                request.manufacturer_serial, "manufacturer_serial"
+            ),
+        }
+        try:
+            with self.repository.transaction() as repository:
+                component = repository.create_component(
+                    component_type=request.component_type,
+                    asset_id=asset_id,
+                    created_by=actor_user_id,
+                    **values,
+                )
+        except DuplicateComponentAssetID as exc:
+            raise DuplicateComponentAssetID(asset_id or "") from exc
+        return self._map_component(component)
+
+    def get_component(self, component_id: UUID) -> ComponentDetail:
+        component = self.repository.get_component(component_id)
+        if component is None:
+            raise ComponentNotFound(component_id)
+        return self._map_component(component)

backend/app/identity.py

@@ -691,7 +691,32 @@ def repair_completion_request(path: str) -> bool:
     )
 
 
+def repair_parts_request(path: str) -> bool:
+    parts = path.split("/")
+    return (
+        len(parts) == 5
+        and parts[:3] == ["", "api", "repairs"]
+        and parts[4] == "parts"
+    ) or (
+        len(parts) == 7
+        and parts[:3] == ["", "api", "repairs"]
+        and parts[4] == "parts"
+        and parts[6] == "void"
+    )
+
+
+def component_request(path: str) -> bool:
+    parts = path.split("/")
+    return path == "/api/components" or (
+        len(parts) == 4 and parts[:3] == ["", "api", "components"]
+    )
+
+
 def required_api_permission(method: str, path: str) -> str | None:
+    if component_request(path):
+        return "components.create" if method == "POST" else "components.read"
+    if repair_parts_request(path):
+        return "repairs.read" if method == "GET" else "repairs.parts.manage"
     if path.startswith("/api/admin/"):
         return "users.manage"
     if method == "POST" and path.endswith("/change-mac") and path.startswith("/api/miners/"):

backend/app/main.py

@@ -19,6 +19,7 @@ from app.identity import (
     router as identity_router,
 )
 from app.inventory.router import router as inventory_router
+from app.components.router import router as components_router
 from app.repairs.exceptions import (
     ControlBoardReplacementActorRequired,
     InvalidMacAddress,
@@ -41,6 +42,7 @@ from app.repairs.router import (
 app = FastAPI(title=settings.app_name)
 app.include_router(identity_router)
 app.include_router(inventory_router)
+app.include_router(components_router)
 install_identity_middleware(app)
 
 

backend/app/repairs/exceptions.py

@@ -135,3 +135,38 @@ class MacOwnershipConflict(ControlBoardReplacementConflict):
         super().__init__(
             "MAC address has current or historical ownership on another miner"
         )
+
+
+class RepairPartNotFound(RepairError):
+    """Raised when a nested repair-part UUID does not exist for a repair."""
+
+    def __init__(self, repair_part_id: object):
+        self.repair_part_id = repair_part_id
+        super().__init__(f"Repair part not found: {repair_part_id}")
+
+
+class RepairPartComponentNotFound(RepairError):
+    """Raised when a tracked component UUID does not exist."""
+
+    def __init__(self, component_id: object):
+        self.component_id = component_id
+        super().__init__(f"Component not found: {component_id}")
+
+
+class RepairPartValidationError(RepairError):
+    """Raised when repair-part input violates a domain rule."""
+
+
+class RepairPartActorRequired(RepairError):
+    """Raised when a repair-part action lacks an authenticated actor."""
+
+    def __init__(self):
+        super().__init__("An authenticated actor is required for repair-part actions")
+
+
+class RepairPartAlreadyVoided(RepairConflict):
+    """Raised when an immutable repair-part entry was already voided."""
+
+    def __init__(self, repair_part_id: object):
+        self.repair_part_id = repair_part_id
+        super().__init__(f"Repair part is already voided: {repair_part_id}")

backend/app/repairs/models.py

@@ -1,10 +1,12 @@
 """Typed contracts for the Repairs domain."""
 
 from datetime import datetime
+from decimal import Decimal
 from typing import Any, Literal
 from uuid import UUID
 
-from pydantic import AwareDatetime, BaseModel, ConfigDict, Field
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, model_validator
+from app.components.models import ComponentType
 
 
 RepairReason = Literal[
@@ -33,6 +35,8 @@ RepairEventType = Literal[
     "INITIAL_NOTE",
     "REPAIR_COMPLETED",
     "CONTROL_BOARD_REPLACED",
+    "REPAIR_PART_RECORDED",
+    "REPAIR_PART_VOIDED",
 ]
 RepairEligibleLifecycle = Literal["active", "spare"]
 RepairPostLifecycle = Literal["active", "spare"]
@@ -194,3 +198,91 @@ class RepairTimelineResponse(BaseModel):
     data: RepairTimeline
     meta: dict[str, object] = Field(default_factory=dict)
     error: None = None
+
+
+ConsumableUnit = Literal["each", "gram", "millilitre", "metre", "centimetre"]
+RepairPartEntryKind = Literal["consumable", "tracked_component"]
+
+
+class ConsumableInput(BaseModel):
+    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
+
+    description: str = Field(min_length=1, max_length=255)
+    quantity: Decimal = Field(gt=0, max_digits=12, decimal_places=3)
+    unit: ConsumableUnit
+
+    @model_validator(mode="after")
+    def validate_each_quantity(self) -> "ConsumableInput":
+        if self.unit == "each" and self.quantity != self.quantity.to_integral_value():
+            raise ValueError("quantity for unit 'each' must be a whole number")
+        return self
+
+
+class RepairPartRecordRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
+
+    consumable: ConsumableInput | None = None
+    tracked_component_uuid: UUID | None = None
+    notes: str | None = Field(default=None, max_length=4000)
+
+    @model_validator(mode="after")
+    def validate_entry_shape(self) -> "RepairPartRecordRequest":
+        if (self.consumable is None) == (self.tracked_component_uuid is None):
+            raise ValueError(
+                "exactly one of consumable and tracked_component_uuid is required"
+            )
+        return self
+
+
+class RepairPartVoidRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
+
+    reason: str = Field(min_length=1, max_length=1000)
+
+
+class RepairPartConsumable(BaseModel):
+    description: str
+    quantity: Decimal
+    unit: ConsumableUnit
+
+
+class RepairPartComponentSnapshot(BaseModel):
+    component_type: ComponentType
+    asset_id: str | None = None
+    manufacturer: str | None = None
+    model: str | None = None
+    manufacturer_serial: str | None = None
+
+
+class RepairPartDetail(BaseModel):
+    repair_part_id: UUID
+    repair_id: UUID
+    entry_kind: RepairPartEntryKind
+    consumable: RepairPartConsumable | None = None
+    component_id: UUID | None = None
+    component_snapshot: RepairPartComponentSnapshot | None = None
+    notes: str | None = None
+    recorded_by: UUID
+    recorded_at: AwareDatetime
+    recorded_event_id: UUID
+    voided_by: UUID | None = None
+    voided_at: AwareDatetime | None = None
+    void_reason: str | None = None
+    void_event_id: UUID | None = None
+
+
+class RepairPartList(BaseModel):
+    repair_id: UUID
+    entries: list[RepairPartDetail] = Field(default_factory=list)
+
+
+class RepairPartResponse(BaseModel):
+    data: RepairPartDetail
+    meta: dict[str, object] = Field(default_factory=dict)
+    error: None = None
+
+
+class RepairPartListResponse(BaseModel):
+    data: RepairPartList
+    meta: dict[str, object] = Field(default_factory=dict)
+    error: None = None

backend/app/repairs/repository.py

@@ -597,6 +597,184 @@ class RepairRepository:
                     row = cur.fetchone()
                 return dict(row)
 
+    def lock_component_for_update(self, component_id: UUID) -> dict | None:
+        with self._connection_scope() as connection:
+            with connection.cursor(row_factory=dict_row) as cur:
+                cur.execute(
+                    """
+                    SELECT
+                        id,
+                        component_type,
+                        asset_id,
+                        manufacturer,
+                        model,
+                        manufacturer_serial
+                    FROM components
+                    WHERE id = %s
+                    FOR UPDATE
+                    """,
+                    (component_id,),
+                )
+                row = cur.fetchone()
+                return dict(row) if row else None
+
+    def create_repair_part(
+        self,
+        *,
+        repair_part_id: UUID,
+        repair_id: UUID,
+        consumable_description: str | None,
+        quantity: object | None,
+        unit: str | None,
+        component: Mapping[str, Any] | None,
+        notes: str | None,
+        recorded_event: Mapping[str, Any],
+    ) -> dict:
+        with self._connection_scope(write=True) as connection:
+            with connection.cursor(row_factory=dict_row) as cur:
+                cur.execute(
+                    """
+                    INSERT INTO repair_parts (
+                        id,
+                        repair_id,
+                        consumable_description,
+                        quantity,
+                        unit,
+                        component_id,
+                        component_type_snapshot,
+                        component_asset_id_snapshot,
+                        component_manufacturer_snapshot,
+                        component_model_snapshot,
+                        component_manufacturer_serial_snapshot,
+                        notes,
+                        recorded_by,
+                        recorded_at,
+                        recorded_event_id
+                    )
+                    VALUES (
+                        %s, %s, %s, %s, %s, %s, %s, %s,
+                        %s, %s, %s, %s, %s, %s, %s
+                    )
+                    RETURNING *
+                    """,
+                    (
+                        repair_part_id,
+                        repair_id,
+                        consumable_description,
+                        quantity,
+                        unit,
+                        component.get("id") if component else None,
+                        component.get("component_type") if component else None,
+                        component.get("asset_id") if component else None,
+                        component.get("manufacturer") if component else None,
+                        component.get("model") if component else None,
+                        component.get("manufacturer_serial") if component else None,
+                        notes,
+                        recorded_event["created_by"],
+                        recorded_event["created_at"],
+                        recorded_event["id"],
+                    ),
+                )
+                return dict(cur.fetchone())
+
+    def get_repair_part(
+        self,
+        repair_id: UUID,
+        repair_part_id: UUID,
+    ) -> dict | None:
+        with self._connection_scope() as connection:
+            with connection.cursor(row_factory=dict_row) as cur:
+                cur.execute(
+                    """
+                    SELECT part.*, event.sequence AS recorded_event_sequence
+                    FROM repair_parts AS part
+                    JOIN repair_events AS event
+                      ON event.id = part.recorded_event_id
+                    WHERE part.repair_id = %s AND part.id = %s
+                    """,
+                    (repair_id, repair_part_id),
+                )
+                row = cur.fetchone()
+                return dict(row) if row else None
+
+    def get_repair_parts(self, repair_id: UUID) -> list[dict] | None:
+        with self._connection_scope() as connection:
+            with connection.cursor(row_factory=dict_row) as cur:
+                cur.execute(
+                    """
+                    SELECT
+                        repair.id AS existing_repair_id,
+                        part.*,
+                        event.sequence AS recorded_event_sequence
+                    FROM repairs AS repair
+                    LEFT JOIN repair_parts AS part
+                      ON part.repair_id = repair.id
+                    LEFT JOIN repair_events AS event
+                      ON event.id = part.recorded_event_id
+                    WHERE repair.id = %s
+                    ORDER BY event.sequence ASC, part.id ASC
+                    """,
+                    (repair_id,),
+                )
+                rows = [dict(row) for row in cur.fetchall()]
+
+        if not rows:
+            return None
+        return [row for row in rows if row.get("id") is not None]
+
+    def lock_repair_part_for_update(
+        self,
+        repair_id: UUID,
+        repair_part_id: UUID,
+    ) -> dict | None:
+        with self._connection_scope() as connection:
+            with connection.cursor(row_factory=dict_row) as cur:
+                cur.execute(
+                    """
+                    SELECT *
+                    FROM repair_parts
+                    WHERE repair_id = %s AND id = %s
+                    FOR UPDATE
+                    """,
+                    (repair_id, repair_part_id),
+                )
+                row = cur.fetchone()
+                return dict(row) if row else None
+
+    def void_repair_part(
+        self,
+        repair_id: UUID,
+        repair_part_id: UUID,
+        *,
+        reason: str,
+        void_event: Mapping[str, Any],
+    ) -> dict | None:
+        with self._connection_scope(write=True) as connection:
+            with connection.cursor(row_factory=dict_row) as cur:
+                cur.execute(
+                    """
+                    UPDATE repair_parts
+                    SET voided_by = %s,
+                        voided_at = %s,
+                        void_reason = %s,
+                        void_event_id = %s
+                    WHERE repair_id = %s
+                      AND id = %s
+                      AND voided_at IS NULL
+                    RETURNING *
+                    """,
+                    (
+                        void_event["created_by"],
+                        void_event["created_at"],
+                        reason,
+                        void_event["id"],
+                        repair_id,
+                        repair_part_id,
+                    ),
+                )
+                row = cur.fetchone()
+                return dict(row) if row else None
+
     def get_repair_detail(self, repair_id: UUID) -> dict | None:
         with self._connection_scope() as connection:
             with connection.cursor(row_factory=dict_row) as cur:

backend/app/repairs/router.py

@@ -15,6 +15,10 @@ from app.repairs.exceptions import (
     RepairConflict,
     RepairMinerNotFound,
     RepairNotFound,
+    RepairPartActorRequired,
+    RepairPartComponentNotFound,
+    RepairPartNotFound,
+    RepairPartValidationError,
 )
 from app.repairs.models import (
     BeginRepairRequest,
@@ -24,6 +28,10 @@ from app.repairs.models import (
     RepairCompleteRequest,
     RepairDetailResponse,
     RepairTimelineResponse,
+    RepairPartListResponse,
+    RepairPartRecordRequest,
+    RepairPartResponse,
+    RepairPartVoidRequest,
 )
 from app.repairs.service import RepairService
 
@@ -33,6 +41,7 @@ control_board_router = APIRouter(prefix="/api/miners", tags=["miners"])
 require_repair_create = require_permission("repairs.create")
 require_repair_read = require_permission("repairs.read")
 require_repair_complete = require_permission("repairs.complete")
+require_repair_parts_manage = require_permission("repairs.parts.manage")
 require_control_board_replace = require_permission("miners.replace_control_board")
 
 RepairServiceDependency = Annotated[
@@ -42,6 +51,7 @@ RepairServiceDependency = Annotated[
 RepairCreateUser = Annotated[dict, Depends(require_repair_create)]
 RepairReadUser = Annotated[dict, Depends(require_repair_read)]
 RepairCompleteUser = Annotated[dict, Depends(require_repair_complete)]
+RepairPartsManageUser = Annotated[dict, Depends(require_repair_parts_manage)]
 ControlBoardReplaceUser = Annotated[dict, Depends(require_control_board_replace)]
 
 
@@ -149,6 +159,107 @@ def complete_repair(
     return RepairDetailResponse(data=repair)
 
 
+@router.get(
+    "/{repair_id}/parts",
+    response_model=RepairPartListResponse,
+)
+def list_repair_parts(
+    repair_id: UUID,
+    service: RepairServiceDependency,
+    current_user: RepairReadUser,
+) -> RepairPartListResponse:
+    del current_user
+    try:
+        parts = service.list_repair_parts(repair_id)
+    except RepairNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Repair not found",
+        ) from exc
+    return RepairPartListResponse(data=parts)
+
+
+@router.post(
+    "/{repair_id}/parts",
+    response_model=RepairPartResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def record_repair_part(
+    repair_id: UUID,
+    request: RepairPartRecordRequest,
+    service: RepairServiceDependency,
+    current_user: RepairPartsManageUser,
+) -> RepairPartResponse:
+    try:
+        part = service.record_repair_part(
+            repair_id,
+            request,
+            actor_user_id=UUID(str(current_user["id"])),
+        )
+    except (RepairNotFound, RepairPartComponentNotFound) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except RepairMinerNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Repair miner no longer exists",
+        ) from exc
+    except RepairConflict as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(exc),
+        ) from exc
+    except (RepairPartActorRequired, RepairPartValidationError) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(exc),
+        ) from exc
+    return RepairPartResponse(data=part)
+
+
+@router.post(
+    "/{repair_id}/parts/{repair_part_id}/void",
+    response_model=RepairPartResponse,
+)
+def void_repair_part(
+    repair_id: UUID,
+    repair_part_id: UUID,
+    request: RepairPartVoidRequest,
+    service: RepairServiceDependency,
+    current_user: RepairPartsManageUser,
+) -> RepairPartResponse:
+    try:
+        part = service.void_repair_part(
+            repair_id,
+            repair_part_id,
+            request,
+            actor_user_id=UUID(str(current_user["id"])),
+        )
+    except (RepairNotFound, RepairPartNotFound) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except RepairMinerNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Repair miner no longer exists",
+        ) from exc
+    except RepairConflict as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(exc),
+        ) from exc
+    except (RepairPartActorRequired, RepairPartValidationError) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(exc),
+        ) from exc
+    return RepairPartResponse(data=part)
+
+
 @router.get(
     "/{repair_id}",
     response_model=RepairDetailResponse,

backend/app/repairs/service.py

@@ -1,6 +1,7 @@
 """Business rules and orchestration for repair workflows."""
 
-from uuid import UUID
+from decimal import Decimal
+from uuid import UUID, uuid4
 
 import re
 
@@ -19,6 +20,11 @@ from app.repairs.exceptions import (
     RepairNotActive,
     RepairNotFound,
     RepairPersistenceError,
+    RepairPartActorRequired,
+    RepairPartAlreadyVoided,
+    RepairPartComponentNotFound,
+    RepairPartNotFound,
+    RepairPartValidationError,
 )
 from app.repairs.models import (
     BeginRepairRequest,
@@ -30,6 +36,12 @@ from app.repairs.models import (
     RepairActor,
     RepairCompleteRequest,
     RepairCompletedPayload,
+    RepairPartComponentSnapshot,
+    RepairPartConsumable,
+    RepairPartDetail,
+    RepairPartList,
+    RepairPartRecordRequest,
+    RepairPartVoidRequest,
     RepairDetail,
     RepairTimeline,
     RepairTimelineEvent,
@@ -378,6 +390,270 @@ class RepairService:
 
         return self._map_repair_detail(detail)
 
+    @staticmethod
+    def _normalize_repair_part_text(
+        value: str | None,
+        *,
+        field_name: str,
+        max_length: int,
+    ) -> str | None:
+        if value is None:
+            return None
+        normalized = str(value).strip()
+        if not normalized:
+            raise RepairPartValidationError(
+                f"{field_name} must not be blank when supplied"
+            )
+        if len(normalized) > max_length:
+            raise RepairPartValidationError(
+                f"{field_name} must not exceed {max_length} characters"
+            )
+        return normalized
+
+    @staticmethod
+    def _validate_consumable_quantity(quantity: Decimal, unit: str) -> Decimal:
+        normalized = Decimal(quantity)
+        if normalized <= 0:
+            raise RepairPartValidationError("quantity must be greater than zero")
+        significant = normalized.normalize()
+        fractional_places = max(-significant.as_tuple().exponent, 0)
+        if fractional_places > 3 or normalized >= Decimal("1000000000"):
+            raise RepairPartValidationError(
+                "quantity must fit NUMERIC(12,3)"
+            )
+        if unit == "each" and normalized != normalized.to_integral_value():
+            raise RepairPartValidationError(
+                "quantity for unit 'each' must be a whole number"
+            )
+        return normalized
+
+    @staticmethod
+    def _map_repair_part(part: dict) -> RepairPartDetail:
+        if part["consumable_description"] is not None:
+            consumable = RepairPartConsumable(
+                description=part["consumable_description"],
+                quantity=part["quantity"],
+                unit=part["unit"],
+            )
+            component_snapshot = None
+            entry_kind = "consumable"
+        else:
+            consumable = None
+            component_snapshot = RepairPartComponentSnapshot(
+                component_type=part["component_type_snapshot"],
+                asset_id=part["component_asset_id_snapshot"],
+                manufacturer=part["component_manufacturer_snapshot"],
+                model=part["component_model_snapshot"],
+                manufacturer_serial=(
+                    part["component_manufacturer_serial_snapshot"]
+                ),
+            )
+            entry_kind = "tracked_component"
+
+        return RepairPartDetail(
+            repair_part_id=part["id"],
+            repair_id=part["repair_id"],
+            entry_kind=entry_kind,
+            consumable=consumable,
+            component_id=part["component_id"],
+            component_snapshot=component_snapshot,
+            notes=part["notes"],
+            recorded_by=part["recorded_by"],
+            recorded_at=part["recorded_at"],
+            recorded_event_id=part["recorded_event_id"],
+            voided_by=part["voided_by"],
+            voided_at=part["voided_at"],
+            void_reason=part["void_reason"],
+            void_event_id=part["void_event_id"],
+        )
+
+    @staticmethod
+    def _lock_active_repair(
+        repository: RepairRepository,
+        repair_id: UUID,
+    ) -> tuple[dict, dict]:
+        identity = repository.get_repair_identity(repair_id)
+        if identity is None:
+            raise RepairNotFound(repair_id)
+
+        miner_id = identity["miner_id"]
+        miner = repository.lock_miner_for_update(miner_id)
+        if miner is None:
+            raise RepairMinerNotFound(miner_id)
+
+        repair = repository.lock_repair_for_update(repair_id)
+        if repair is None:
+            raise RepairNotFound(repair_id)
+        if repair["miner_id"] != miner_id:
+            raise RepairActiveMismatch(repair_id, miner_id)
+        if repair["status"] not in ACTIVE_REPAIR_STATUSES:
+            raise RepairNotActive(repair_id, repair["status"])
+
+        active_repair = repository.get_active_repair(miner_id)
+        if active_repair is None or active_repair["id"] != repair_id:
+            raise RepairActiveMismatch(repair_id, miner_id)
+        lifecycle_status = str(miner["lifecycle_status"])
+        if lifecycle_status != REPAIR_LIFECYCLE_STATUS:
+            raise RepairMinerNotInRepair(miner_id, lifecycle_status)
+        return miner, repair
+
+    def record_repair_part(
+        self,
+        repair_id: UUID,
+        request: RepairPartRecordRequest,
+        *,
+        actor_user_id: UUID | None,
+    ) -> RepairPartDetail:
+        if actor_user_id is None:
+            raise RepairPartActorRequired()
+
+        notes = self._normalize_repair_part_text(
+            request.notes,
+            field_name="notes",
+            max_length=4000,
+        )
+        description = None
+        quantity = None
+        unit = None
+        if request.consumable is not None:
+            description = self._normalize_repair_part_text(
+                request.consumable.description,
+                field_name="consumable.description",
+                max_length=255,
+            )
+            quantity = self._validate_consumable_quantity(
+                request.consumable.quantity,
+                request.consumable.unit,
+            )
+            unit = request.consumable.unit
+
+        with self.repository.transaction() as repository:
+            self._lock_active_repair(repository, repair_id)
+
+            component = None
+            if request.tracked_component_uuid is not None:
+                component = repository.lock_component_for_update(
+                    request.tracked_component_uuid
+                )
+                if component is None:
+                    raise RepairPartComponentNotFound(
+                        request.tracked_component_uuid
+                    )
+
+            repair_part_id = uuid4()
+            next_sequence = repository.get_next_event_sequence(repair_id)
+            if component is None:
+                payload = {
+                    "repair_part_id": str(repair_part_id),
+                    "entry_kind": "consumable",
+                    "description": description,
+                    "quantity": format(quantity, "f"),
+                    "unit": unit,
+                    "notes": notes,
+                }
+            else:
+                payload = {
+                    "repair_part_id": str(repair_part_id),
+                    "entry_kind": "tracked_component",
+                    "component_id": str(component["id"]),
+                    "component_type": component["component_type"],
+                    "component_asset_id": component["asset_id"],
+                    "component_manufacturer": component["manufacturer"],
+                    "component_model": component["model"],
+                    "component_manufacturer_serial": (
+                        component["manufacturer_serial"]
+                    ),
+                    "notes": notes,
+                }
+            event = repository.create_event(
+                repair_id,
+                sequence=next_sequence,
+                event_type="REPAIR_PART_RECORDED",
+                created_by=actor_user_id,
+                payload=payload,
+            )
+            repository.create_repair_part(
+                repair_part_id=repair_part_id,
+                repair_id=repair_id,
+                consumable_description=description,
+                quantity=quantity,
+                unit=unit,
+                component=component,
+                notes=notes,
+                recorded_event=event,
+            )
+            part = repository.get_repair_part(repair_id, repair_part_id)
+            if part is None:
+                raise RepairPersistenceError(
+                    "Recorded repair part could not be read in its transaction"
+                )
+
+        return self._map_repair_part(part)
+
+    def list_repair_parts(self, repair_id: UUID) -> RepairPartList:
+        parts = self.repository.get_repair_parts(repair_id)
+        if parts is None:
+            raise RepairNotFound(repair_id)
+        return RepairPartList(
+            repair_id=repair_id,
+            entries=[self._map_repair_part(part) for part in parts],
+        )
+
+    def void_repair_part(
+        self,
+        repair_id: UUID,
+        repair_part_id: UUID,
+        request: RepairPartVoidRequest,
+        *,
+        actor_user_id: UUID | None,
+    ) -> RepairPartDetail:
+        if actor_user_id is None:
+            raise RepairPartActorRequired()
+        reason = self._normalize_repair_part_text(
+            request.reason,
+            field_name="reason",
+            max_length=1000,
+        )
+        assert reason is not None
+
+        with self.repository.transaction() as repository:
+            self._lock_active_repair(repository, repair_id)
+            part = repository.lock_repair_part_for_update(
+                repair_id,
+                repair_part_id,
+            )
+            if part is None:
+                raise RepairPartNotFound(repair_part_id)
+            if part["voided_at"] is not None:
+                raise RepairPartAlreadyVoided(repair_part_id)
+
+            next_sequence = repository.get_next_event_sequence(repair_id)
+            event = repository.create_event(
+                repair_id,
+                sequence=next_sequence,
+                event_type="REPAIR_PART_VOIDED",
+                created_by=actor_user_id,
+                payload={
+                    "repair_part_id": str(repair_part_id),
+                    "reason": reason,
+                },
+            )
+            voided = repository.void_repair_part(
+                repair_id,
+                repair_part_id,
+                reason=reason,
+                void_event=event,
+            )
+            if voided is None:
+                raise RepairPartAlreadyVoided(repair_part_id)
+            part = repository.get_repair_part(repair_id, repair_part_id)
+            if part is None:
+                raise RepairPersistenceError(
+                    "Voided repair part could not be read in its transaction"
+                )
+
+        return self._map_repair_part(part)
+
     def get_repair(self, repair_id: UUID) -> RepairDetail:
         repair = self.repository.get_repair_detail(repair_id)
         if repair is None:

backend/tests/test_control_board_replacement.py

@@ -474,7 +474,9 @@ class ControlBoardReplacementDatabaseTests(unittest.TestCase):
                     "TRUNCATE miner_control_board_installations, "
                     "miner_network_identities, miner_mac_ownerships"
                 )
+                cur.execute("ALTER TABLE repair_events DISABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repair_events")
+                cur.execute("ALTER TABLE repair_events ENABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repairs")
                 cur.execute("DELETE FROM miner_tags")
                 cur.execute("DELETE FROM miner_asset_id_history")
@@ -500,7 +502,9 @@ class ControlBoardReplacementDatabaseTests(unittest.TestCase):
                     "TRUNCATE miner_control_board_installations, "
                     "miner_network_identities, miner_mac_ownerships"
                 )
+                cur.execute("ALTER TABLE repair_events DISABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repair_events")
+                cur.execute("ALTER TABLE repair_events ENABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repairs")
                 cur.execute("DELETE FROM miner_tags")
                 cur.execute("DELETE FROM miner_asset_id_history")

backend/tests/test_repair_parts_tracking.py

@@ -0,0 +1,455 @@
+import ast
+from contextlib import contextmanager
+from copy import deepcopy
+from datetime import datetime, timezone
+from decimal import Decimal
+from pathlib import Path
+import unittest
+from uuid import uuid4
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from pydantic import ValidationError
+
+from app.components.dependencies import get_component_service
+from app.components.exceptions import ComponentValidationError
+from app.components.models import ComponentCreateRequest, ComponentDetail
+from app.components.router import router as component_router
+from app.components.service import ComponentService
+from app.identity import get_current_user, required_api_permission
+from app.repairs.dependencies import get_repair_service
+from app.repairs.exceptions import RepairPartAlreadyVoided
+from app.repairs.models import (
+    ConsumableInput,
+    RepairPartComponentSnapshot,
+    RepairPartConsumable,
+    RepairPartDetail,
+    RepairPartList,
+    RepairPartRecordRequest,
+    RepairPartVoidRequest,
+)
+from app.repairs.router import router as repair_router
+from app.repairs.service import RepairService
+
+
+NOW = datetime(2026, 8, 2, 12, 0, tzinfo=timezone.utc)
+
+
+class FakeComponentRepository:
+    def __init__(self):
+        self.rows = {}
+        self.calls = []
+
+    @contextmanager
+    def transaction(self):
+        self.calls.append("transaction")
+        yield self
+
+    def create_component(self, **values):
+        component_id = uuid4()
+        row = {
+            "component_id": component_id,
+            "created_at": NOW,
+            **values,
+        }
+        self.rows[component_id] = row
+        return row
+
+    def get_component(self, component_id):
+        return self.rows.get(component_id)
+
+
+class ComponentServiceTests(unittest.TestCase):
+    def test_create_normalizes_metadata_and_uses_authoritative_fields(self):
+        repository = FakeComponentRepository()
+        actor = uuid4()
+        component = ComponentService(repository).create_component(
+            ComponentCreateRequest(
+                component_type="hashboard",
+                asset_id="  HB-0042  ",
+                manufacturer="  MicroBT  ",
+            ),
+            actor_user_id=actor,
+        )
+        self.assertEqual(component.asset_id, "HB-0042")
+        self.assertEqual(component.manufacturer, "MicroBT")
+        self.assertEqual(component.created_by, actor)
+        self.assertIsNotNone(component.component_id)
+        self.assertEqual(repository.calls, ["transaction"])
+
+    def test_component_request_rejects_blank_unknown_and_server_fields(self):
+        for body, exception in (
+            ({"component_type": "other"}, ValidationError),
+            ({"component_type": "fan", "asset_id": "   "}, ComponentValidationError),
+            ({"component_type": "fan", "created_by": str(uuid4())}, ValidationError),
+        ):
+            with self.subTest(body=body), self.assertRaises(exception):
+                request = ComponentCreateRequest(**body)
+                ComponentService(FakeComponentRepository()).create_component(
+                    request, actor_user_id=uuid4()
+                )
+
+
+class FakeRepairPartRepository:
+    def __init__(self):
+        self.repair_id = uuid4()
+        self.miner_id = uuid4()
+        self.actor_id = uuid4()
+        self.component_id = uuid4()
+        self.miner = {"id": self.miner_id, "lifecycle_status": "in_repair"}
+        self.repair = {
+            "id": self.repair_id,
+            "miner_id": self.miner_id,
+            "status": "OPEN",
+        }
+        self.component = {
+            "id": self.component_id,
+            "component_type": "control_board",
+            "asset_id": "CB-42",
+            "manufacturer": "MicroBT",
+            "model": None,
+            "manufacturer_serial": "SER-42",
+        }
+        self.events = [{"sequence": 1}]
+        self.parts = {}
+        self.calls = []
+        self.fail_at = None
+
+    @contextmanager
+    def transaction(self):
+        self.calls.append("transaction")
+        snapshot = deepcopy((self.events, self.parts))
+        try:
+            yield self
+        except Exception:
+            self.events, self.parts = snapshot
+            raise
+
+    def _call(self, name):
+        self.calls.append(name)
+        if self.fail_at == name:
+            raise RuntimeError(name)
+
+    def get_repair_identity(self, repair_id):
+        self._call("repair_identity")
+        return {"miner_id": self.miner_id} if repair_id == self.repair_id else None
+
+    def lock_miner_for_update(self, miner_id):
+        self._call("lock_miner")
+        return self.miner if miner_id == self.miner_id else None
+
+    def lock_repair_for_update(self, repair_id):
+        self._call("lock_repair")
+        return self.repair if repair_id == self.repair_id else None
+
+    def get_active_repair(self, miner_id):
+        self._call("active_repair")
+        return self.repair if miner_id == self.miner_id else None
+
+    def lock_component_for_update(self, component_id):
+        self._call("lock_component")
+        return self.component if component_id == self.component_id else None
+
+    def get_next_event_sequence(self, repair_id):
+        self._call("next_sequence")
+        return max(event["sequence"] for event in self.events) + 1
+
+    def create_event(self, repair_id, **values):
+        self._call("create_event")
+        row = {"id": uuid4(), "repair_id": repair_id, "created_at": NOW, **values}
+        self.events.append(row)
+        return row
+
+    def create_repair_part(self, **values):
+        self._call("create_part")
+        component = values["component"]
+        row = {
+            "id": values["repair_part_id"],
+            "repair_id": values["repair_id"],
+            "consumable_description": values["consumable_description"],
+            "quantity": values["quantity"],
+            "unit": values["unit"],
+            "component_id": component["id"] if component else None,
+            "component_type_snapshot": component["component_type"] if component else None,
+            "component_asset_id_snapshot": component["asset_id"] if component else None,
+            "component_manufacturer_snapshot": component["manufacturer"] if component else None,
+            "component_model_snapshot": component["model"] if component else None,
+            "component_manufacturer_serial_snapshot": component["manufacturer_serial"] if component else None,
+            "notes": values["notes"],
+            "recorded_by": values["recorded_event"]["created_by"],
+            "recorded_at": values["recorded_event"]["created_at"],
+            "recorded_event_id": values["recorded_event"]["id"],
+            "voided_by": None,
+            "voided_at": None,
+            "void_reason": None,
+            "void_event_id": None,
+        }
+        self.parts[row["id"]] = row
+        return row
+
+    def get_repair_part(self, repair_id, repair_part_id):
+        self._call("get_part")
+        row = self.parts.get(repair_part_id)
+        return row if row and row["repair_id"] == repair_id else None
+
+    def get_repair_parts(self, repair_id):
+        rows = [row for row in self.parts.values() if row["repair_id"] == repair_id]
+        return rows if repair_id == self.repair_id else None
+
+    def lock_repair_part_for_update(self, repair_id, repair_part_id):
+        self._call("lock_part")
+        return self.get_repair_part(repair_id, repair_part_id)
+
+    def void_repair_part(self, repair_id, repair_part_id, *, reason, void_event):
+        self._call("void_part")
+        row = self.get_repair_part(repair_id, repair_part_id)
+        if row is None or row["voided_at"] is not None:
+            return None
+        row.update(
+            voided_by=void_event["created_by"],
+            voided_at=void_event["created_at"],
+            void_reason=reason,
+            void_event_id=void_event["id"],
+        )
+        return row
+
+
+class RepairPartServiceTests(unittest.TestCase):
+    def setUp(self):
+        self.repository = FakeRepairPartRepository()
+        self.service = RepairService(self.repository)
+
+    def test_request_xor_precision_units_and_trim(self):
+        valid = ConsumableInput(description=" Paste ", quantity="1.250", unit="gram")
+        self.assertEqual(valid.description, "Paste")
+        self.assertEqual(valid.quantity, Decimal("1.250"))
+        invalid = (
+            {"consumable": None, "tracked_component_uuid": None},
+            {"consumable": valid, "tracked_component_uuid": uuid4()},
+            {"consumable": {"description": "x", "quantity": "1.0001", "unit": "gram"}},
+            {"consumable": {"description": "x", "quantity": "1.5", "unit": "each"}},
+            {"consumable": {"description": "x", "quantity": "1", "unit": "kg"}},
+        )
+        for body in invalid:
+            with self.subTest(body=body), self.assertRaises(ValidationError):
+                RepairPartRecordRequest(**body)
+
+    def test_consumable_record_is_atomic_and_sequenced(self):
+        result = self.service.record_repair_part(
+            self.repository.repair_id,
+            RepairPartRecordRequest(
+                consumable={"description": " Paste ", "quantity": "2", "unit": "gram"},
+                notes=" Used on board ",
+            ),
+            actor_user_id=self.repository.actor_id,
+        )
+        self.assertEqual(result.entry_kind, "consumable")
+        self.assertEqual(result.consumable.description, "Paste")
+        event = self.repository.events[-1]
+        self.assertEqual(event["event_type"], "REPAIR_PART_RECORDED")
+        self.assertEqual(event["sequence"], 2)
+        self.assertEqual(event["created_by"], self.repository.actor_id)
+        self.assertEqual(event["payload"]["quantity"], "2")
+        self.assertEqual(self.repository.calls[:5], [
+            "transaction", "repair_identity", "lock_miner", "lock_repair", "active_repair"
+        ])
+
+    def test_tracked_component_uses_locked_snapshot_without_mutation(self):
+        before = deepcopy(self.repository.component)
+        result = self.service.record_repair_part(
+            self.repository.repair_id,
+            RepairPartRecordRequest(tracked_component_uuid=self.repository.component_id),
+            actor_user_id=self.repository.actor_id,
+        )
+        self.assertEqual(result.component_id, self.repository.component_id)
+        self.assertEqual(result.component_snapshot.asset_id, "CB-42")
+        self.assertEqual(self.repository.component, before)
+        self.assertLess(
+            self.repository.calls.index("lock_repair"),
+            self.repository.calls.index("lock_component"),
+        )
+
+    def test_void_is_one_way_and_rollbacks_leave_no_partial_event(self):
+        part = self.service.record_repair_part(
+            self.repository.repair_id,
+            RepairPartRecordRequest(
+                consumable={"description": "Fuse", "quantity": 1, "unit": "each"}
+            ),
+            actor_user_id=self.repository.actor_id,
+        )
+        voided = self.service.void_repair_part(
+            self.repository.repair_id,
+            part.repair_part_id,
+            RepairPartVoidRequest(reason=" Wrong repair "),
+            actor_user_id=self.repository.actor_id,
+        )
+        self.assertEqual(voided.void_reason, "Wrong repair")
+        with self.assertRaises(RepairPartAlreadyVoided):
+            self.service.void_repair_part(
+                self.repository.repair_id,
+                part.repair_part_id,
+                RepairPartVoidRequest(reason="Again"),
+                actor_user_id=self.repository.actor_id,
+            )
+        self.assertEqual(
+            [event.get("event_type") for event in self.repository.events].count("REPAIR_PART_VOIDED"),
+            1,
+        )
+
+    def test_insert_failure_rolls_back_event_and_part(self):
+        self.repository.fail_at = "create_part"
+        with self.assertRaises(RuntimeError):
+            self.service.record_repair_part(
+                self.repository.repair_id,
+                RepairPartRecordRequest(
+                    consumable={"description": "Fuse", "quantity": 1, "unit": "each"}
+                ),
+                actor_user_id=self.repository.actor_id,
+            )
+        self.assertEqual(self.repository.events, [{"sequence": 1}])
+        self.assertEqual(self.repository.parts, {})
+
+
+def part_detail(repair_id, actor_id):
+    return RepairPartDetail(
+        repair_part_id=uuid4(), repair_id=repair_id, entry_kind="consumable",
+        consumable=RepairPartConsumable(description="Fuse", quantity=1, unit="each"),
+        component_id=None, component_snapshot=None, notes=None,
+        recorded_by=actor_id, recorded_at=NOW, recorded_event_id=uuid4(),
+    )
+
+
+class StubComponentService:
+    def __init__(self, actor):
+        self.actor = actor
+
+    def create_component(self, request, *, actor_user_id):
+        self.actor_seen = actor_user_id
+        return ComponentDetail(
+            component_id=uuid4(), component_type=request.component_type,
+            created_at=NOW, created_by=actor_user_id,
+        )
+
+    def get_component(self, component_id):
+        return ComponentDetail(
+            component_id=component_id, component_type="fan",
+            created_at=NOW, created_by=self.actor,
+        )
+
+
+class StubRepairService:
+    def __init__(self, repair_id, actor):
+        self.repair_id = repair_id
+        self.actor = actor
+        self.detail = part_detail(repair_id, actor)
+
+    def record_repair_part(self, repair_id, request, *, actor_user_id):
+        self.actor_seen = actor_user_id
+        return self.detail
+
+    def list_repair_parts(self, repair_id):
+        return RepairPartList(repair_id=repair_id, entries=[self.detail])
+
+    def void_repair_part(self, repair_id, repair_part_id, request, *, actor_user_id):
+        self.actor_seen = actor_user_id
+        return self.detail.model_copy(update={
+            "voided_by": actor_user_id, "voided_at": NOW,
+            "void_reason": request.reason, "void_event_id": uuid4(),
+        })
+
+
+class Task005RouterTests(unittest.TestCase):
+    def setUp(self):
+        self.actor = uuid4()
+        self.repair_id = uuid4()
+        self.component_service = StubComponentService(self.actor)
+        self.repair_service = StubRepairService(self.repair_id, self.actor)
+        self.app = FastAPI()
+        self.app.include_router(component_router)
+        self.app.include_router(repair_router)
+        self.app.dependency_overrides[get_component_service] = lambda: self.component_service
+        self.app.dependency_overrides[get_repair_service] = lambda: self.repair_service
+        self.client = TestClient(self.app)
+
+    def tearDown(self):
+        self.client.close()
+
+    def authenticate(self, permissions, role="Technician"):
+        self.app.dependency_overrides[get_current_user] = lambda: {
+            "id": str(self.actor), "role": role, "permissions": list(permissions)
+        }
+
+    def test_authentication_permissions_and_actor_authority(self):
+        path = f"/api/repairs/{self.repair_id}/parts"
+        body = {"consumable": {"description": "Fuse", "quantity": 1, "unit": "each"}}
+        self.assertEqual(self.client.post(path, json=body).status_code, 401)
+        self.authenticate(("repairs.read",), role="Read Only")
+        self.assertEqual(self.client.post(path, json=body).status_code, 403)
+        self.assertEqual(self.client.get(path).status_code, 200)
+        self.authenticate(("repairs.parts.manage",))
+        self.assertEqual(self.client.post(path, json=body).status_code, 201)
+        self.assertEqual(self.repair_service.actor_seen, self.actor)
+
+    def test_component_routes_permissions_and_envelopes(self):
+        self.authenticate(("components.read",), role="Read Only")
+        self.assertEqual(self.client.post("/api/components", json={"component_type": "fan"}).status_code, 403)
+        self.assertEqual(self.client.get(f"/api/components/{uuid4()}").status_code, 200)
+        self.authenticate(("components.create",))
+        response = self.client.post("/api/components", json={"component_type": "fan"})
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(response.json()["meta"], {})
+        self.assertIsNone(response.json()["error"])
+
+    def test_exact_routes_and_permission_mapping(self):
+        component_routes = {(route.path, tuple(sorted(route.methods))) for route in component_router.routes}
+        self.assertEqual(component_routes, {
+            ("/api/components", ("POST",)),
+            ("/api/components/{component_id}", ("GET",)),
+        })
+        part_routes = {
+            (route.path, tuple(sorted(route.methods)))
+            for route in repair_router.routes if "/parts" in route.path
+        }
+        self.assertEqual(part_routes, {
+            ("/api/repairs/{repair_id}/parts", ("GET",)),
+            ("/api/repairs/{repair_id}/parts", ("POST",)),
+            ("/api/repairs/{repair_id}/parts/{repair_part_id}/void", ("POST",)),
+        })
+        repair_id = uuid4()
+        part_id = uuid4()
+        self.assertEqual(required_api_permission("GET", "/api/components"), "components.read")
+        self.assertEqual(required_api_permission("POST", "/api/components"), "components.create")
+        self.assertEqual(required_api_permission("GET", f"/api/repairs/{repair_id}/parts"), "repairs.read")
+        self.assertEqual(required_api_permission("POST", f"/api/repairs/{repair_id}/parts"), "repairs.parts.manage")
+        self.assertEqual(required_api_permission("POST", f"/api/repairs/{repair_id}/parts/{part_id}/void"), "repairs.parts.manage")
+
+    def test_architecture_boundaries(self):
+        root = Path(__file__).parents[1] / "app"
+        for relative in ("components/router.py", "repairs/router.py"):
+            source = (root / relative).read_text(encoding="utf-8")
+            tree = ast.parse(source)
+            imports = [
+                node.module for node in ast.walk(tree)
+                if isinstance(node, ast.ImportFrom) and node.module
+            ]
+            self.assertFalse(any(name.startswith(("psycopg", "sqlalchemy")) for name in imports))
+            self.assertNotIn("SELECT ", source.upper())
+            self.assertNotIn(".COMMIT(", source.upper())
+        for relative in (
+            "components/service.py", "components/repository.py",
+            "repairs/service.py", "repairs/repository.py",
+        ):
+            tree = ast.parse((root / relative).read_text(encoding="utf-8"))
+            imports = []
+            for node in ast.walk(tree):
+                if isinstance(node, ast.Import):
+                    imports.extend(alias.name for alias in node.names)
+                elif isinstance(node, ast.ImportFrom) and node.module:
+                    imports.append(node.module)
+            self.assertFalse(
+                any(name == "fastapi" or name.startswith("fastapi.") for name in imports)
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_repair_timeline.py

@@ -219,7 +219,9 @@ class RepairTimelineRepositoryTests(unittest.TestCase):
         self.note_actor = uuid4()
         with self.repository._connection_scope(write=True) as connection:
             with connection.cursor() as cur:
+                cur.execute("ALTER TABLE repair_events DISABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repair_events")
+                cur.execute("ALTER TABLE repair_events ENABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repairs")
                 cur.execute("DELETE FROM miner_tags")
                 cur.execute("DELETE FROM miner_asset_id_history")
@@ -611,7 +613,7 @@ class RepairTimelineRouterTests(unittest.TestCase):
         self.assertEqual(response.status_code, 403)
         self.assertFalse(self.service.begin_called)
 
-    def test_exactly_two_timeline_get_routes_are_registered(self):
+    def test_timeline_and_parts_get_routes_are_registered(self):
         get_routes = {
             route.path
             for route in router.routes
@@ -622,6 +624,7 @@ class RepairTimelineRouterTests(unittest.TestCase):
             {
                 "/api/repairs/{repair_id}",
                 "/api/repairs/{repair_id}/events",
+                "/api/repairs/{repair_id}/parts",
             },
         )
 
@@ -715,6 +718,21 @@ class RepairTimelineArchitectureTests(unittest.TestCase):
                     "minerops_repair_parts_summary_api",
                 ),
                 ("POST", "/api/repairs", "begin_repair"),
+                (
+                    "GET",
+                    "/api/repairs/{repair_id}/parts",
+                    "list_repair_parts",
+                ),
+                (
+                    "POST",
+                    "/api/repairs/{repair_id}/parts",
+                    "record_repair_part",
+                ),
+                (
+                    "POST",
+                    "/api/repairs/{repair_id}/parts/{repair_part_id}/void",
+                    "void_repair_part",
+                ),
                 (
                     "POST",
                     "/api/repairs/{repair_id}/complete",

backend/tests/test_repairs_domain.py

@@ -372,7 +372,9 @@ class RepairRepositoryIntegrationTests(unittest.TestCase):
                 cur.execute(
                     "DROP FUNCTION IF EXISTS test_fail_repair_lifecycle()"
                 )
+                cur.execute("ALTER TABLE repair_events DISABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repair_events")
+                cur.execute("ALTER TABLE repair_events ENABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repairs")
                 cur.execute("DELETE FROM miner_tags")
                 cur.execute("DELETE FROM miner_asset_id_history")

backend/tests/test_return_to_service.py

@@ -417,7 +417,9 @@ class ReturnToServiceRepositoryTests(unittest.TestCase):
                 cur.execute(
                     "DROP FUNCTION IF EXISTS test_fail_completion_lifecycle()"
                 )
+                cur.execute("ALTER TABLE repair_events DISABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repair_events")
+                cur.execute("ALTER TABLE repair_events ENABLE TRIGGER trg_protect_repair_event_history")
                 cur.execute("DELETE FROM repairs")
                 cur.execute("DELETE FROM miner_tags")
                 cur.execute("DELETE FROM miner_asset_id_history")
@@ -586,7 +588,14 @@ class ReturnToServiceRepositoryTests(unittest.TestCase):
                 _, repair_id = self.begin()
                 with self.repository._connection_scope(write=True) as connection:
                     with connection.cursor() as cur:
+                        if name == "delete":
+                            cur.execute(
+                                "ALTER TABLE repair_events DISABLE TRIGGER "
+                                "trg_protect_repair_event_history"
+                            )
                         cur.execute(change_sql, (repair_id,))
+                        if name == "delete":
+                            cur.execute("ALTER TABLE repair_events ENABLE TRIGGER trg_protect_repair_event_history")
 
                 RepairService(self.repository).complete_repair(
                     repair_id,