Add token-gated SDLC pipeline approvals

orchestrator/api.py

@@ -48,6 +48,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
     from routes.metrics import metrics_bp
     from routes.phases import phases_bp
     from routes.pipelines import pipelines_bp
+    from routes.sdlc_tokens import sdlc_tokens_bp
     from routes.signals import signals_bp
     from webhooks import webhooks_bp
 
@@ -59,13 +60,15 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
     app.register_blueprint(decisions_bp)
     app.register_blueprint(metrics_bp)
     app.register_blueprint(webhooks_bp)
+    app.register_blueprint(sdlc_tokens_bp)
 except ImportError:
     from .routes.containers import containers_bp  # type: ignore[no-redef]
     from .routes.decisions import decisions_bp  # type: ignore[no-redef]
     from .routes.health import health_bp  # type: ignore[no-redef]
     from .routes.metrics import metrics_bp  # type: ignore[no-redef]
     from .routes.phases import phases_bp  # type: ignore[no-redef]
     from .routes.pipelines import pipelines_bp  # type: ignore[no-redef]
+    from .routes.sdlc_tokens import sdlc_tokens_bp  # type: ignore[no-redef]
     from .routes.signals import signals_bp  # type: ignore[no-redef]
     from .webhooks import webhooks_bp  # type: ignore[no-redef]
 
@@ -77,6 +80,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
     app.register_blueprint(decisions_bp)
     app.register_blueprint(metrics_bp)
     app.register_blueprint(webhooks_bp)
+    app.register_blueprint(sdlc_tokens_bp)
 
 
 @app.before_request

orchestrator/models.py

@@ -207,6 +207,9 @@ class Pipeline(BaseModel):
     )
     updated_at: datetime = Field(default_factory=datetime.utcnow, description="Last update time")
     contract_synced: bool = Field(default=True, description="Whether state is synced with contract")
+    sdlc_token_gated: bool = Field(
+        default=False, description="Whether this pipeline requires token-gated approval"
+    )
     error: str | None = Field(default=None, description="Error if failed")
     version: int = Field(
         default=1, ge=1, description="Optimistic locking version (incremented on each save)"

orchestrator/routes/decisions.py

@@ -35,7 +35,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
     DecisionNotFoundError,
     get_decision_queue,
 )
-from state_store import InvalidPipelineIdError
+from state_store import InvalidPipelineIdError, PipelineNotFoundError, get_state_store
 
 logger = get_logger("orchestrator.decisions")
 
@@ -287,6 +287,21 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int]
     if not resolution:
         return make_error_response("Missing resolution")
 
+    # Check if pipeline is token-gated — block direct resolution if so
+    try:
+        store = get_state_store(repo_path)
+        pipeline = store.load_pipeline(pipeline_id)
+        if pipeline.sdlc_token_gated:
+            return make_error_response(
+                "This pipeline requires token-gated approval. "
+                "Type !approve <phase> in your terminal.",
+                status_code=403,
+            )
+    except PipelineNotFoundError:
+        pass  # Pipeline may not exist yet, allow resolution
+    except Exception:
+        pass  # Don't block on state store errors
+
     try:
         queue = get_decision_queue(pipeline_id, repo_path)
         decision = queue.resolve_decision(decision_id, resolution)

orchestrator/routes/pipelines.py

@@ -435,6 +435,9 @@ def create_pipeline() -> tuple[Response, int]:
     repo = data.get("repo")
     branch = data.get("branch")
 
+    # Check for pre-generated SDLC tokens (set by entrypoint before Claude starts)
+    from routes.sdlc_tokens import has_tokens_for_pipeline
+
     if not issue_number:
         return make_error_response("Missing issue_number")
     if not repo:
@@ -479,6 +482,12 @@ def create_pipeline() -> tuple[Response, int]:
                 status_code=500,
             )
         pipeline.contract_synced = True
+
+        # Enable token gating if tokens were pre-generated for this pipeline
+        if has_tokens_for_pipeline(pipeline.id):
+            pipeline.sdlc_token_gated = True
+            logger.info("Pipeline token-gated", pipeline_id=pipeline.id)
+
         store.save_pipeline(pipeline, commit=False)
         logger.info(
             "Issue pipeline contract created",

orchestrator/routes/sdlc_tokens.py

@@ -0,0 +1,218 @@
+"""
+SDLC token-gated approval endpoints.
+
+Provides token generation and validation for SDLC pipeline phase approvals.
+Tokens are stored in-memory (ephemeral, single-session lifetime).
+"""
+
+import hashlib
+import secrets
+import sys
+from pathlib import Path
+from typing import Any
+
+from flask import Blueprint, Response, jsonify, request
+
+# Add parent directory to path for imports
+_parent_path = Path(__file__).parent.parent
+if str(_parent_path) not in sys.path:
+    sys.path.insert(0, str(_parent_path))
+
+# Add shared directory to path for logging
+_shared_path = Path(__file__).parent.parent.parent / "shared"
+if _shared_path.exists() and str(_shared_path) not in sys.path:
+    sys.path.insert(0, str(_shared_path))
+
+try:
+    from egg_logging import get_logger
+except ImportError:
+    import logging
+
+    def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
+        return logging.getLogger(name)
+
+
+from sdlc_wordlist import WORD_LIST
+
+logger = get_logger("orchestrator.sdlc_tokens")
+
+sdlc_tokens_bp = Blueprint("sdlc_tokens", __name__, url_prefix="/api/v1/sdlc-tokens")
+
+# In-memory token store: pipeline_id -> token data
+# Acceptable because tokens are ephemeral (one session lifetime),
+# orchestrator is a singleton, restart invalidates tokens.
+_token_store: dict[str, dict[str, Any]] = {}
+
+VALID_PHASES = {"refine", "plan"}
+
+
+def _generate_token() -> str:
+    """Generate a 3-word token like APPLE-HORSE-RIVER."""
+    words = [secrets.choice(WORD_LIST) for _ in range(3)]
+    return "-".join(words)
+
+
+def _hash_token(token: str) -> str:
+    """SHA-256 hash a token for storage."""
+    return hashlib.sha256(token.upper().encode()).hexdigest()
+
+
+def has_tokens_for_pipeline(pipeline_id: str) -> bool:
+    """Check if tokens have been generated for a pipeline."""
+    return pipeline_id in _token_store
+
+
+def _make_error(message: str, status_code: int = 400) -> tuple[Response, int]:
+    """Create an error response."""
+    return jsonify({"success": False, "message": message}), status_code
+
+
+def _make_success(message: str, data: dict[str, Any] | None = None) -> tuple[Response, int]:
+    """Create a success response."""
+    response: dict[str, Any] = {"success": True, "message": message}
+    if data:
+        response["data"] = data
+    return jsonify(response), 200
+
+
+@sdlc_tokens_bp.route("/generate", methods=["POST"])
+def generate_tokens() -> tuple[Response, int]:
+    """Generate approval tokens for an SDLC pipeline.
+
+    Request body:
+        {"pipeline_id": "issue-596"}
+
+    Returns plaintext tokens (one-time display to human).
+    """
+    data = request.get_json() or {}
+    pipeline_id = data.get("pipeline_id")
+
+    if not pipeline_id:
+        return _make_error("Missing pipeline_id")
+
+    if pipeline_id in _token_store:
+        return _make_error(f"Tokens already generated for pipeline {pipeline_id}", 409)
+
+    refine_token = _generate_token()
+    plan_token = _generate_token()
+
+    # Ensure tokens are different
+    while plan_token == refine_token:
+        plan_token = _generate_token()
+
+    _token_store[pipeline_id] = {
+        "refine_hash": _hash_token(refine_token),
+        "plan_hash": _hash_token(plan_token),
+        "refine_used": False,
+        "plan_used": False,
+    }
+
+    logger.info("SDLC tokens generated", pipeline_id=pipeline_id)
+
+    return _make_success(
+        "Tokens generated",
+        data={
+            "pipeline_id": pipeline_id,
+            "refine_token": refine_token,
+            "plan_token": plan_token,
+        },
+    )
+
+
+@sdlc_tokens_bp.route("/approve", methods=["POST"])
+def approve_phase() -> tuple[Response, int]:
+    """Validate a token and approve an SDLC phase.
+
+    Request body:
+        {
+            "pipeline_id": "issue-596",
+            "phase": "refine",
+            "token": "APPLE-HORSE-RIVER"
+        }
+
+    Returns:
+        200: Phase approved
+        400: Bad input
+        403: Wrong token
+        404: No tokens for pipeline
+        409: Token already used
+    """
+    data = request.get_json() or {}
+    pipeline_id = data.get("pipeline_id")
+    phase = data.get("phase")
+    token = data.get("token")
+
+    if not pipeline_id:
+        return _make_error("Missing pipeline_id")
+    if not phase:
+        return _make_error("Missing phase")
+    if not token:
+        return _make_error("Missing token")
+
+    if phase not in VALID_PHASES:
+        return _make_error(f"Invalid phase: {phase}. Must be one of: {', '.join(sorted(VALID_PHASES))}")
+
+    if pipeline_id not in _token_store:
+        return _make_error(f"No tokens found for pipeline {pipeline_id}", 404)
+
+    store = _token_store[pipeline_id]
+    hash_key = f"{phase}_hash"
+    used_key = f"{phase}_used"
+
+    if store[used_key]:
+        return _make_error(f"Token for phase '{phase}' has already been used", 409)
+
+    # Timing-safe comparison of token hashes
+    provided_hash = _hash_token(token)
+    if not secrets.compare_digest(provided_hash, store[hash_key]):
+        logger.warning(
+            "SDLC token validation failed",
+            pipeline_id=pipeline_id,
+            phase=phase,
+        )
+        return _make_error("Invalid token", 403)
+
+    # Mark token as used
+    store[used_key] = True
+
+    # Resolve any pending decisions for this pipeline/phase
+    _resolve_phase_decisions(pipeline_id, phase)
+
+    logger.info("SDLC phase approved", pipeline_id=pipeline_id, phase=phase)
+
+    return _make_success(
+        f"Phase '{phase}' approved",
+        data={"pipeline_id": pipeline_id, "phase": phase},
+    )
+
+
+def _resolve_phase_decisions(pipeline_id: str, phase: str) -> None:
+    """Resolve pending HITL decisions for a pipeline phase."""
+    from routes import get_repo_path
+
+    try:
+        from decision_queue import get_decision_queue
+
+        repo_path = get_repo_path()
+        queue = get_decision_queue(pipeline_id, repo_path)
+        pending = queue.get_pending_decisions()
+
+        for decision in pending:
+            # Match decisions that reference this phase
+            question_lower = decision.question.lower()
+            if phase in question_lower:
+                queue.resolve_decision(decision.id, f"Approved via SDLC token ({phase})")
+                logger.info(
+                    "Auto-resolved decision via SDLC token",
+                    pipeline_id=pipeline_id,
+                    decision_id=decision.id,
+                    phase=phase,
+                )
+    except Exception as e:
+        # Don't fail the approval if decision resolution fails
+        logger.warning(
+            "Failed to auto-resolve decisions",
+            pipeline_id=pipeline_id,
+            phase=phase,
+            error=str(e),
+        )

orchestrator/sdlc_wordlist.py

@@ -0,0 +1,54 @@
+"""
+Word list for SDLC approval token generation.
+
+Contains ~200 concrete, unambiguous, easy-to-spell English nouns (3-7 chars each).
+3-word tokens yield 200^3 = 8M combinations, sufficient for ephemeral single-session use.
+"""
+
+WORD_LIST = [
+    # Animals
+    "BEAR", "BIRD", "CAT", "CRAB", "CROW", "DEER", "DOG", "DOVE", "DUCK",
+    "EAGLE", "EEL", "ELK", "FISH", "FOX", "FROG", "GOAT", "GOOSE", "HAWK",
+    "HARE", "HORSE", "LAMB", "LION", "MOLE", "MOOSE", "MOTH", "MOUSE",
+    "NEWT", "OWL", "PANDA", "PIG", "PONY", "RAM", "ROBIN", "SEAL", "SHARK",
+    "SHEEP", "SLUG", "SNAIL", "SNAKE", "STORK", "SWAN", "TIGER", "TOAD",
+    "TROUT", "VIPER", "WASP", "WHALE", "WOLF", "WORM", "ZEBRA",
+    # Colors
+    "AMBER", "BLACK", "BLUE", "BROWN", "CORAL", "CREAM", "GOLD", "GREEN",
+    "GREY", "IVORY", "LILAC", "NAVY", "OLIVE", "PEACH", "PINK", "PLUM",
+    "RED", "RUBY", "RUST", "TAN", "TEAL", "WHITE",
+    # Fruits & food
+    "APPLE", "BASIL", "BEAN", "BERRY", "BREAD", "CANDY", "CEDAR", "CHERRY",
+    "CHILI", "COCOA", "DATE", "FIG", "GRAPE", "GUAVA", "HONEY", "KIWI",
+    "LEMON", "LIME", "MANGO", "MAPLE", "MELON", "MINT", "OLIVE", "PEACH",
+    "PEAR", "PLUM", "RICE", "SAGE", "WHEAT",
+    # Weather & sky
+    "BOLT", "BREEZE", "CLOUD", "DEW", "FLAME", "FLOOD", "FOG", "FROST",
+    "GALE", "HAIL", "ICE", "MIST", "MOON", "RAIN", "SKY", "SLEET",
+    "SNOW", "STAR", "STORM", "SUN", "TIDE", "WIND",
+    # Nature & landscape
+    "ASH", "BAY", "BLUFF", "BOG", "BROOK", "CAPE", "CAVE", "CLIFF",
+    "COAST", "COVE", "CREEK", "DALE", "DELTA", "DUNE", "FALLS", "FIELD",
+    "FJORD", "GLEN", "GROVE", "GULF", "HILL", "ISLE", "LAKE", "MARSH",
+    "MESA", "MOSS", "OAK", "PALM", "PEAK", "PINE", "POND", "REEF",
+    "RIDGE", "RIVER", "ROCK", "SAND", "SHORE", "SLOPE", "STONE", "TRAIL",
+    "VALE", "VINE", "WOOD",
+    # Tools & objects
+    "ANVIL", "AXE", "BELL", "BOLT", "BOW", "BRICK", "BRUSH", "CHAIN",
+    "CHEST", "CLOCK", "COIN", "CROWN", "DRUM", "FLUTE", "FORGE", "GATE",
+    "GEAR", "GLASS", "GLOBE", "GONG", "HARP", "HELM", "HORN", "KEY",
+    "KNOT", "LANCE", "LATCH", "LENS", "LOCK", "MASK", "NAIL", "OPAL",
+    "PEARL", "PIKE", "PIPE", "PRISM", "RING", "ROPE", "SCALE", "SHIELD",
+    "SPEAR", "SPIKE", "SWORD", "TORCH", "TOWER", "VAULT", "WAGON",
+    "WHEEL",
+]
+
+# Deduplicate while preserving order
+_seen: set[str] = set()
+_unique: list[str] = []
+for _w in WORD_LIST:
+    if _w not in _seen:
+        _seen.add(_w)
+        _unique.append(_w)
+WORD_LIST = _unique
+del _seen, _unique, _w