Parser confidence as a deployment gate

Introduction to Staged Workflow

A staged workflow splits a monolithic extraction‑to‑action pipeline into discrete, observable stages. Each stage has a defined transaction boundary (pre‑conditions, commit point, verification gate). If verification fails, the workflow can trigger a rollback and surface an operator intervention point. Treating each confidence‑based branch as its own transaction limits blast‑radius, makes failure modes explicit, and gives operators a clear place to pause, inspect, and correct before automation proceeds.

Extraction models rarely produce deterministic outputs; they emit a confidence score. Treating every extraction as “truth” introduces silent data corruption that can cascade into downstream systems. Mapping confidence bands to distinct actions creates a risk‑aware execution boundary:

• Low confidence → manual verification (operator intervention) → prevents bad data from entering downstream systems.
• Medium confidence → automated clarification (additional commands) → reduces reliance on human effort while gathering evidence.
• High confidence → automated push → only after a verification gate confirms the extraction meets pre‑defined correctness criteria.

This approach transforms an opaque, all‑or‑nothing model into a series of checkpoints that can be measured, alerted on, and rolled back if needed.

Designing the Staged Workflow

Low‑Confidence Extractions and Human Review

Triggering Human Review

1. Extraction service emits JSON { "value": "...", "confidence": 0.0‑1.0 }.
2. If confidence < LOW_THRESHOLD (e.g., 0.6), route payload to a human‑review queue.
3. No further automated processing occurs until a human acknowledges the item.
4. Reviewer either accepts (verified = true) or rejects (verified = false, optionally providing a corrected value).
5. On rejection, discard the original low‑confidence extraction and write a compensating log entry; no downstream state change occurs.
6. The review UI presents the raw source, model confidence, and contextual evidence; the reviewer must explicitly confirm before the item leaves the queue.

Integrating Human Review Feedback

• Accepted items are re‑tagged with confidence = 1.0 (or a high‑confidence placeholder) and forwarded to the medium‑confidence stage.
• Every review action (accept/reject, timestamp, reviewer ID, optional comment) is written to an immutable audit store (e.g., append‑only Kafka topic or write‑once S3 object).
• On rejection, a dead‑letter queue captures the original extraction for later model‑retraining analysis.
• Because the human‑review stage is a synchronous, single‑consumer queue (or limited‑size worker pool), a surge of low‑confidence items cannot overwhelm downstream automation; back‑pressure is applied at the queue level.

Medium‑Confidence Results and Additional Commands

Requesting Additional Commands for Clarification

1. Payloads with LOW_THRESHOLD ≤ confidence < HIGH_THRESHOLD (e.g., 0.6‑0.85) enter the clarification stage.
2. The stage issues one or more targeted commands (e.g., CLI show interface, SNMP GET, or API GET /resource/{id}) designed to raise confidence.
3. Responses are collected, but no state change occurs until the verification gate evaluates the aggregated evidence.
4. A rule‑based or lightweight ML scorer combines the original extraction with the new command output to compute an updated confidence. If the updated confidence ≥ HIGH_THRESHOLD, the item is promoted; otherwise it is sent back for another round (up to a configurable retry limit) or downgraded to low confidence for human review.
5. If clarification commands fail (timeout, error) and retries are exhausted, the item is failed‑fast and sent to an error queue for operator inspection.
6. After the maximum retry count, an alert is raised; an operator can manually decide to accept the original extraction, request a different command, or discard the item.

Handling Responses to Additional Commands

• Each command type has a dedicated parser that converts raw output (text, JSON, XML) into a normalized feature vector.
• Features are concatenated with the original extraction features; a logistic regression or decision‑tree model outputs a new confidence score.
• Sending the same clarification command multiple times is safe (read‑only); duplicate responses are treated as additional evidence, not as state‑changing actions.
• Clarification commands are directed at specific resources (e.g., a single device interface); per‑resource rate limits prevent overwhelming devices.

High‑Confidence State and Automation/Analytics Pipelines

Defining Verification Criteria for High‑Confidence State

1. Items with confidence ≥ HIGH_THRESHOLD (e.g., ≥0.85) are considered candidates for automation.
2. Before pipeline ingestion, run a deterministic verification:
   • Schema validation (JSON‑Schema or Avro).
   • Cross‑field consistency checks (e.g., extracted IP must belong to a known subnet).
   • Optional out‑of‑band verification (e.g., quick ping or ARP lookup) that is idempotent and low‑cost.
3. If all verification passes, the item is committed to the downstream pipeline; otherwise it is rerouted to the medium‑confidence stage for further clarification.
4. Should the downstream pipeline detect an inconsistency after ingestion (e.g., duplicate key violation), it emits a compensating event that triggers a rollback workflow: the offending record is marked invalid, a dead‑letter entry is created, and an alert is raised for operator review.
5. The rollback workflow includes a manual approval step before purging or correcting the bad record, ensuring automated rollbacks do not silently delete data.

Integrating with Automation and Analytics Pipelines

• Automation: High‑confidence events are published to a command‑execution topic (e.g., Kafka automation.commands). Consumers are idempotent automation agents (Ansible playbooks, Terraform applies, or custom Python controllers) that execute only if the event’s event_id has not been processed before (checked via a deduplication store).
• Analytics: A parallel stream feeds into an analytics lake (e.g., S3 + Athena or Snowflake). The lake receives the raw extraction plus the final confidence flag, enabling downstream data‑quality metrics.
• Exactly‑once semantics: The pipeline uses transactional producers (Kafka) or idempotent writes (S3 multipart with ETag checks) to guarantee that a single extraction does not result in duplicate automation actions or duplicate analytics rows.
• Blast‑radius: Automation consumers are limited to a concurrency ceiling (e.g., 10 workers) and operate behind a circuit breaker that trips if error rates exceed a threshold, preventing a runaway change from affecting many devices simultaneously.

Implementing the Staged Workflow

Choosing the Right Technologies and Tools

Workflow state store
Table workflow_items: item_id UUID PK, source_data JSONB, confidence REAL, stage TEXT, verification JSONB, updated_at TIMESTAMP. Indexes on stage and confidence enable fast queue polling.

Audit log
Append‑only Kafka topic workflow.audit: each message {event_type, item_id, timestamp, user_id, payload_hash}. Kafka log compaction retains a latest snapshot for replay.

Dead‑letter / retry queues
Separate Kafka topics workflow.dlq.low, workflow.dlq.medium, workflow.dlq.high with configurable retention (e.g., 7 days) for forensic analysis.

Object store for large artifacts
Raw command outputs (e.g., CLI logs) stored in S3 with metadata item_id and stage; lifecycle rules transition to Glacier after 30 days for cost efficiency.

Building the Workflow Components

Low‑Confidence Extraction Handling (Python/Temporal)

# low_confidence_handler.py
import json
import logging
from datetime import timedelta
from temporalio import activity, workflow
from db import get_session, WorkflowItem
from audit import log_audit

LOW_THRESHOLD = 0.6

@activity.defn
async def route_to_human_review(item_id: str) -> None:
    """Move item to human‑review queue and audit the transition."""
    async with get_session() as session:
        item = await session.get(WorkflowItem, item_id)
        if item.confidence < LOW_THRESHOLD:
            item.stage = "human_review"
            await session.commit()
            await log_audit(
                event_type="stage_transition",
                item_id=item_id,
                payload={"from": item.stage, "to": "human_review"},
            )
        else:
            raise ValueError(f"Item {item_id} confidence {item.confidence} >= low threshold")

@workflow.defn
class HumanReviewWorkflow:
    @workflow.run
    async def run(self, item_id: str):
        await workflow.execute_activity(
            route_to_human_review,
            item_id,
            start_to_close_timeout=timedelta(seconds=30),
        )
        # Wait for human signal via a signal channel
        decision = await workflow.wait_condition(
            lambda: workflow.info().is_signalled("human_decision")
        )
        # Human decision signal carries {accepted: bool, corrected_value?: str}
        if decision.accepted:
            # Promote to medium confidence stage
            await workflow.execute_activity(
                promote_to_medium,
                item_id,
                start_to_close_timeout=timedelta(seconds=30),
            )
        else:
            # Send to dead‑letter for model retraining
            await workflow.execute_activity(
                send_to_dlq,
                item_id,
                reason="human_reject",
                start_to_close_timeout=timedelta(seconds=30),
            )

The workflow explicitly defines a transaction boundary (route_to_human_review), a verification gate (human decision signal), and a rollback path (dead‑letter).

Medium‑Confidence Clarification (Bash)

# medium_confidence_clarify.sh
#!/usr/bin/env bash
set -euo pipefail

ITEM_ID=$1
LOW_THRESH=0.6
HIGH_THRESH=0.85
MAX_RETRIES=2

# Fetch item from PostgreSQL
read -r source_data confidence <<<$(psql "dbname=workflow user=app" -t -c \
    "SELECT source_data, confidence FROM workflow_items WHERE item_id='$ITEM_ID';")

# If already low confidence, abort – should be handled elsewhere
if (( $(echo "$confidence < $LOW_THRESH" | bc -l) )); then
    echo "Item $ITEM_ID is low confidence; should be in human review."
    exit 1
fi

# If already high confidence, skip clarification
if (( $(echo "$confidence >= $HIGH_THRESH" | bc -l) )); then
    echo "Item $ITEM_ID already high confidence; skipping clarification."
    exit 0
fi

retry=0
while [ $retry -le $MAX_RETRIES ]; do
    # Example CLI command via Netmiko wrapper (Python inline)
    output=$(python3 - <<PY
from netmiko import ConnectHandler
device = {
    'device_type': 'cisco_ios',
    'host': '10.0.0.5',
    'username': 'admin',
    'password': 'secret',
}
conn = ConnectHandler(**device)
result = conn.send_command('show interfaces description')
conn.disconnect()
print(result)
PY
    )
    # In a real implementation, parse $output and update confidence.
    # For demonstration, assume we raise confidence after first attempt.
    new_conf=0.9
    if (( $(echo "$new_conf >= $HIGH_THRESH" | bc -l) )); then
        echo "Clarification succeeded; new confidence $new_conf."
        # Update DB (pseudo‑code)
        psql "dbname=workflow user=app" -c \
            "UPDATE workflow_items SET confidence=$new_conf, stage='high_confidence' WHERE item_id='$ITEM_ID';"
        exit 0
    fi
    ((retry++))
    echo "Attempt $retry failed; retrying..."
done

echo "Max retries exceeded; sending to error queue."
# Push to error queue (pseudo‑code)
# kafka-producer --topic workflow.dlq.medium --key $ITEM_ID --value "{\"reason\":\"clarification_failed\"}"
exit 1

The script checks confidence, runs a clarification command, updates confidence on success, and retries up to MAX_RETRIES. On failure, it routes the item to an error queue.

High‑Confidence Verification Sketch (Python)

# high_confidence_verifier.py
import json
from db import get_session, WorkflowItem
from audit import log_audit
from validators import validate_schema, check_subnet, ping_host

HIGH_THRESH = 0.85

async def verify_high_confidence(item_id: str) -> bool:
    async with get_session() as session:
        item = await session.get(WorkflowItem, item_id)
        if item.confidence < HIGH_THRESH:
            return False

        payload = json.loads(item.source_data)
        # Deterministic checks
        if not validate_schema(payload):
            await log_audit("schema_fail", item_id, {})
            return False
        if not check_subnet(payload.get("ip")):
            await log_audit("subnet_fail", item_id, {})
            return False
        if not ping_host(payload.get("ip")):   # idempotent, low‑cost
            await log_audit("ping_fail", item_id, {})
            return False

        # All checks passed – commit to downstream pipeline
        item.stage = "ready_for_automation"
        await session.commit()
        await log_audit("high_conf_verified", item_id, {"confidence": item.confidence})
        return True

This function represents the verification gate before automation/analytics ingestion.