Spec Kitty

└─ kitty-specs
   └─ Glossary Semantic Integrity Runtime for Mission Framework

Mission Run:

📚 Docs ↗

Contracts

events.md

Event Contracts: Glossary Semantic Integrity

Feature: 041-mission-glossary-semantic-integrity Date: 2026-02-16 Status: Complete Authority: Feature 007 (spec-kitty-events package)

Overview

This document references the canonical glossary event schemas defined in Feature 007. CLI implementation MUST import these events from spec-kitty-events package, NOT redefine them locally.

Package import path: spec_kitty_events.glossary.events

Contract authority: /Users/robert/ClaudeCowork/Spec-Kitty-Cowork/spec-kitty-planning/product-ideas/prd-mission-glossary-semantic-integrity-v1.md

Canonical Events (Feature 007)

1. GlossaryScopeActivated

Trigger: Mission starts or scope selected Purpose: Record which glossary scopes are active for this mission run

Required fields:

{
    "event_type": "GlossaryScopeActivated",
    "scope_id": str,              # e.g., "team_domain"
    "glossary_version_id": str,   # e.g., "v3"
    "timestamp": datetime,
    "mission_id": str,
    "run_id": str
}

Consumers: Orchestration runtime, projection layer

Example:

GlossaryScopeActivated(
    scope_id="team_domain",
    glossary_version_id="v3",
    timestamp=datetime(2026, 2, 16, 12, 0, 0),
    mission_id="041-mission",
    run_id="run-2026-02-16-001"
)

2. TermCandidateObserved

Trigger: New or uncertain term appears in step input/output Purpose: Record term extraction for candidate glossary entry

Required fields:

{
    "event_type": "TermCandidateObserved",
    "term": str,                # e.g., "workspace"
    "source_step": str,         # e.g., "step-specify-001"
    "actor_id": str,            # e.g., "user:alice" or "llm:claude-sonnet-4"
    "confidence": float,        # 0.0-1.0
    "extraction_method": str,   # e.g., "metadata_hint", "casing_pattern"
    "context": str,             # e.g., "description field"
    "timestamp": datetime,
    "mission_id": str,
    "run_id": str
}

Consumers: Candidate scorer, clarification policy

Example:

TermCandidateObserved(
    term="workspace",
    source_step="step-specify-001",
    actor_id="llm:claude-sonnet-4",
    confidence=0.8,
    extraction_method="casing_pattern",
    context="description field",
    timestamp=datetime(2026, 2, 16, 12, 5, 0),
    mission_id="041-mission",
    run_id="run-2026-02-16-001"
)

3. SemanticCheckEvaluated

Trigger: Step-level pre-generation validation completes Purpose: Record conflict detection results and recommended action

Required fields:

{
    "event_type": "SemanticCheckEvaluated",
    "step_id": str,
    "mission_id": str,
    "run_id": str,
    "timestamp": datetime,
    "findings": List[ConflictFinding],  # See structure below
    "overall_severity": str,            # "low" | "medium" | "high"
    "confidence": float,                # 0.0-1.0
    "effective_strictness": str,        # "off" | "medium" | "max"
    "recommended_action": str,          # "proceed" | "warn" | "block"
    "blocked": bool                     # True if generation was blocked
}

ConflictFinding structure:

{
    "term": str,
    "conflict_type": str,  # "unknown" | "ambiguous" | "inconsistent" | "unresolved_critical"
    "severity": str,       # "low" | "medium" | "high"
    "confidence": float,
    "candidate_senses": List[SenseRef],  # See structure below
    "context": str
}

SenseRef structure:

{
    "surface": str,
    "scope": str,
    "definition": str,
    "confidence": float
}

Consumers: Execution gate, UX prompt, dashboards

Example:

SemanticCheckEvaluated(
    step_id="step-specify-001",
    mission_id="041-mission",
    run_id="run-2026-02-16-001",
    timestamp=datetime(2026, 2, 16, 12, 10, 0),
    findings=[
        ConflictFinding(
            term="workspace",
            conflict_type="ambiguous",
            severity="high",
            confidence=0.9,
            candidate_senses=[
                SenseRef(surface="workspace", scope="team_domain",
                         definition="Git worktree directory", confidence=0.9),
                SenseRef(surface="workspace", scope="team_domain",
                         definition="VS Code workspace file", confidence=0.7)
            ],
            context="description field"
        )
    ],
    overall_severity="high",
    confidence=0.9,
    effective_strictness="medium",
    recommended_action="block",
    blocked=True
)

4. GlossaryClarificationRequested

Trigger: Policy requires user clarification (high severity or low confidence critical term) Purpose: Record clarification question sent to user

Required fields:

{
    "event_type": "GlossaryClarificationRequested",
    "question": str,          # Human-readable question
    "term": str,              # Term requiring clarification
    "options": List[str],     # Ranked candidate definitions
    "urgency": str,           # "high" | "medium" | "low"
    "timestamp": datetime,
    "mission_id": str,
    "run_id": str,
    "step_id": str,
    "conflict_id": str        # UUID for tracking resolution
}

Consumers: Mission participant UI, CLI, SaaS decision inbox

Example:

GlossaryClarificationRequested(
    question="What does 'workspace' mean in this context?",
    term="workspace",
    options=[
        "Git worktree directory for a work package",
        "VS Code workspace configuration file"
    ],
    urgency="high",
    timestamp=datetime(2026, 2, 16, 12, 15, 0),
    mission_id="041-mission",
    run_id="run-2026-02-16-001",
    step_id="step-specify-001",
    conflict_id="uuid-1234-5678"
)

5. GlossaryClarificationResolved

Trigger: User/participant answer accepted Purpose: Record conflict resolution (selected from candidates)

Required fields:

{
    "event_type": "GlossaryClarificationResolved",
    "conflict_id": str,           # UUID from GlossaryClarificationRequested
    "term_surface": str,          # e.g., "workspace"
    "selected_sense": SenseRef,   # See structure in SemanticCheckEvaluated
    "actor": ActorIdentity,       # See structure below
    "timestamp": datetime,
    "resolution_mode": str,       # "interactive" | "async"
    "provenance": Provenance      # See structure below
}

ActorIdentity structure:

{
    "actor_id": str,
    "actor_type": str,  # "human" | "llm" | "service"
    "display_name": str
}

Provenance structure:

{
    "source": str,      # e.g., "user_clarification"
    "timestamp": datetime,
    "actor_id": str
}

Consumers: Glossary updater, execution gate

Example:

GlossaryClarificationResolved(
    conflict_id="uuid-1234-5678",
    term_surface="workspace",
    selected_sense=SenseRef(
        surface="workspace",
        scope="team_domain",
        definition="Git worktree directory for a work package",
        confidence=0.9
    ),
    actor=ActorIdentity(
        actor_id="user:alice",
        actor_type="human",
        display_name="Alice"
    ),
    timestamp=datetime(2026, 2, 16, 12, 20, 0),
    resolution_mode="interactive",
    provenance=Provenance(
        source="user_clarification",
        timestamp=datetime(2026, 2, 16, 12, 20, 0),
        actor_id="user:alice"
    )
)

6. GlossarySenseUpdated

Trigger: New sense or sense edit accepted (custom definition from user) Purpose: Record glossary modification

Required fields:

{
    "event_type": "GlossarySenseUpdated",
    "term_surface": str,
    "scope": str,                # e.g., "team_domain"
    "new_sense": TermSense,      # See structure below
    "actor": ActorIdentity,
    "timestamp": datetime,
    "update_type": str,          # "create" | "update"
    "provenance": Provenance
}

TermSense structure:

{
    "surface": str,
    "scope": str,
    "definition": str,
    "confidence": float,
    "status": str  # "draft" | "active" | "deprecated"
}

Consumers: Projection, audit views

Example:

GlossarySenseUpdated(
    term_surface="workspace",
    scope="team_domain",
    new_sense=TermSense(
        surface="workspace",
        scope="team_domain",
        definition="Git worktree directory for a work package",
        confidence=1.0,
        status="active"
    ),
    actor=ActorIdentity(actor_id="user:alice", actor_type="human", display_name="Alice"),
    timestamp=datetime(2026, 2, 16, 12, 25, 0),
    update_type="create",
    provenance=Provenance(
        source="user_clarification",
        timestamp=datetime(2026, 2, 16, 12, 25, 0),
        actor_id="user:alice"
    )
)

7. GenerationBlockedBySemanticConflict

Trigger: High-severity unresolved conflict at generation boundary Purpose: Record generation gate block decision

Required fields:

{
    "event_type": "GenerationBlockedBySemanticConflict",
    "step_id": str,
    "mission_id": str,
    "run_id": str,
    "timestamp": datetime,
    "conflicts": List[ConflictFinding],  # See structure in SemanticCheckEvaluated
    "strictness_mode": str,              # "off" | "medium" | "max"
    "effective_strictness": str          # Resolved strictness (after precedence)
}

Consumers: Execution runtime, dashboards, audit logs

Example:

GenerationBlockedBySemanticConflict(
    step_id="step-specify-001",
    mission_id="041-mission",
    run_id="run-2026-02-16-001",
    timestamp=datetime(2026, 2, 16, 12, 30, 0),
    conflicts=[
        ConflictFinding(
            term="workspace",
            conflict_type="ambiguous",
            severity="high",
            confidence=0.9,
            candidate_senses=[...],
            context="description field"
        )
    ],
    strictness_mode="medium",
    effective_strictness="medium"
)

CLI-Specific Events (Pending Feature 007 Approval)

8. StepCheckpointed

Trigger: Before generation gate evaluation (for resume capability) Purpose: Save minimal state for deterministic resume after conflict resolution

Required fields:

{
    "event_type": "StepCheckpointed",
    "mission_id": str,
    "run_id": str,
    "step_id": str,
    "strictness": str,           # "off" | "medium" | "max"
    "scope_refs": List[ScopeRef],  # See structure below
    "input_hash": str,           # SHA256 of step inputs
    "cursor": str,               # e.g., "pre_generation_gate"
    "retry_token": str,          # UUID
    "timestamp": datetime
}

ScopeRef structure:

{
    "scope": str,          # e.g., "team_domain"
    "version_id": str      # e.g., "v3"
}

Consumers: Resume middleware, replay engine

Example:

StepCheckpointed(
    mission_id="041-mission",
    run_id="run-2026-02-16-001",
    step_id="step-specify-001",
    strictness="medium",
    scope_refs=[
        ScopeRef(scope="mission_local", version_id="v1"),
        ScopeRef(scope="team_domain", version_id="v3")
    ],
    input_hash="abc123...",
    cursor="pre_generation_gate",
    retry_token="uuid-9999-0000",
    timestamp=datetime(2026, 2, 16, 12, 35, 0)
)

Status: This event may need to be added to Feature 007 canonical events. If not present in spec-kitty-events package, CLI should stub adapter and gate implementation on package update.

Event Emission Guidelines

1. Import from spec-kitty-events package:

``python from spec_kitty_events.glossary.events import ( GlossaryScopeActivated, TermCandidateObserved, SemanticCheckEvaluated, # ... etc ) ``

2. DO NOT redefine event schemas locally - use canonical contracts from package

3. If event not in package yet (e.g., StepCheckpointed):

Stub adapter boundary in src/specify_cli/glossary/events.py
Document as "pending Feature 007 approval"
Gate implementation on package availability

4. Event order matters for replay:

Lamport clocks ensure causal ordering
Events are append-only, immutable
Replay: same events → same state (deterministic)

5. All events MUST include:

timestamp (ISO 8601 format)
mission_id and run_id (for scoping)
Appropriate actor/provenance metadata

Testing Event Contracts

Unit tests (src/specify_cli/glossary/tests/test_events.py):

Verify event emission at correct middleware boundaries
Validate event payload conforms to schema (all required fields present)
Test event ordering (extraction → check → gate → clarification → resolution)

Integration tests:

Full pipeline: extract → check → block → clarify → resolve → resume
Verify events emitted in correct order
Replay events, assert glossary state matches

Contract tests (if spec-kitty-events package available):

Import canonical event classes
Assert CLI payloads match package schemas
Test serialization/deserialization round-trip

middleware.md

Middleware Contracts: Glossary Semantic Integrity

Feature: 041-mission-glossary-semantic-integrity Date: 2026-02-16 Status: Complete

Overview

This document defines the middleware pipeline interfaces for glossary semantic integrity checks. All middleware components follow a common protocol and execute in a fixed order.

Base Middleware Protocol

All glossary middleware components implement this protocol:

from typing import Protocol
from specify_cli.missions.primitives import PrimitiveExecutionContext

class GlossaryMiddleware(Protocol):
    """Base protocol for glossary middleware components."""

    def process(self, context: PrimitiveExecutionContext) -> PrimitiveExecutionContext:
        """
        Process primitive execution context.

        Args:
            context: Current execution context with inputs, metadata, config

        Returns:
            Modified context (may add fields like extracted_terms, conflicts)

        Raises:
            BlockedByConflict: If generation must be blocked
        """
        ...

Middleware Pipeline Order

Middleware executes in this fixed order:

1. GlossaryCandidateExtractionMiddleware - Extract terms from step I/O 2. SemanticCheckMiddleware - Resolve terms, detect conflicts 3. GenerationGateMiddleware - Block generation on high-severity conflicts 4. ClarificationMiddleware - Prompt user for resolution (if blocked) 5. ResumeMiddleware - Load checkpoint, restore state (on retry)

1. GlossaryCandidateExtractionMiddleware

Purpose: Extract candidate terms from step inputs/outputs using metadata hints + deterministic heuristics.

Interface:

class GlossaryCandidateExtractionMiddleware:
    """Extract candidate terms from step inputs/outputs."""

    def __init__(self, extraction_config: ExtractionConfig):
        """
        Initialize extraction middleware.

        Args:
            extraction_config: Configuration for term extraction
                - metadata_hints: glossary_watch_terms, glossary_aliases, etc.
                - heuristic_patterns: quoted_phrases, acronyms, casing
                - confidence_thresholds: high/medium/low boundaries
        """
        self.extraction_config = extraction_config

    def process(self, context: PrimitiveExecutionContext) -> PrimitiveExecutionContext:
        """
        Extract terms from step inputs/outputs.

        Modifies context:
            - Adds context.extracted_terms: List[ExtractedTerm]

        Emits events:
            - TermCandidateObserved (for each extracted term)

        Returns:
            Modified context with extracted_terms field
        """
        ...

ExtractedTerm structure:

from dataclasses import dataclass

@dataclass
class ExtractedTerm:
    surface: str                # e.g., "workspace"
    confidence: float           # 0.0-1.0
    extraction_method: str      # e.g., "metadata_hint", "casing_pattern"
    context: str                # e.g., "description field"
    source_step: str            # e.g., "step-specify-001"

Example usage:

extraction_middleware = GlossaryCandidateExtractionMiddleware(
    extraction_config=ExtractionConfig(
        metadata_hints={
            "glossary_watch_terms": ["workspace", "mission"],
            "glossary_fields": ["description", "requirements"]
        },
        heuristic_patterns=["quoted_phrases", "acronyms", "casing"],
        confidence_thresholds={"high": 0.8, "medium": 0.5, "low": 0.3}
    )
)

context = extraction_middleware.process(context)
# context.extracted_terms = [ExtractedTerm(...), ...]

2. SemanticCheckMiddleware

Purpose: Resolve extracted terms against scope hierarchy, detect conflicts.

Interface:

class SemanticCheckMiddleware:
    """Resolve terms and detect semantic conflicts."""

    def __init__(self, glossary_store: GlossaryStore):
        """
        Initialize semantic check middleware.

        Args:
            glossary_store: Access to active glossary scopes
                - Provides scope resolution (mission_local -> team_domain -> audience_domain -> spec_kitty_core)
                - Loads seed files from .kittify/glossaries/
        """
        self.glossary_store = glossary_store

    def process(self, context: PrimitiveExecutionContext) -> PrimitiveExecutionContext:
        """
        Resolve terms and detect conflicts.

        Requires:
            - context.extracted_terms (from extraction middleware)

        Modifies context:
            - Adds context.conflicts: List[SemanticConflict]

        Emits events:
            - SemanticCheckEvaluated (with findings, overall severity, recommended action)

        Returns:
            Modified context with conflicts field
        """
        ...

SemanticConflict structure (see data-model.md):

from dataclasses import dataclass
from enum import Enum

class ConflictType(Enum):
    UNKNOWN = "unknown"
    AMBIGUOUS = "ambiguous"
    INCONSISTENT = "inconsistent"
    UNRESOLVED_CRITICAL = "unresolved_critical"

class Severity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

@dataclass
class SemanticConflict:
    term: str                       # e.g., "workspace"
    conflict_type: ConflictType
    severity: Severity
    confidence: float               # 0.0-1.0
    candidate_senses: List[SenseRef]
    context: str                    # Usage location

Example usage:

semantic_check_middleware = SemanticCheckMiddleware(
    glossary_store=GlossaryStore(
        scopes=[GlossaryScope.MISSION_LOCAL, GlossaryScope.TEAM_DOMAIN,
                GlossaryScope.AUDIENCE_DOMAIN, GlossaryScope.SPEC_KITTY_CORE]
    )
)

context = semantic_check_middleware.process(context)
# context.conflicts = [SemanticConflict(...), ...]

3. GenerationGateMiddleware

Purpose: Block LLM generation on unresolved high-severity conflicts.

Interface:

class GenerationGateMiddleware:
    """Block generation on unresolved high-severity conflicts."""

    def __init__(self, strictness_policy: StrictnessPolicy):
        """
        Initialize generation gate middleware.

        Args:
            strictness_policy: Strictness mode configuration
                - Resolves precedence: global -> mission -> step -> runtime override
                - Modes: off, medium, max
        """
        self.strictness_policy = strictness_policy

    def process(self, context: PrimitiveExecutionContext) -> PrimitiveExecutionContext:
        """
        Block generation if high-severity conflicts exist.

        Requires:
            - context.conflicts (from semantic check middleware)
            - context.strictness (resolved strictness mode)

        Emits events:
            - GenerationBlockedBySemanticConflict (if blocked)

        Raises:
            BlockedByConflict: If generation must be blocked

        Returns:
            Unmodified context (if pass)
        """
        ...

Strictness policy logic:

from enum import Enum

class Strictness(Enum):
    OFF = "off"       # No blocking
    MEDIUM = "medium" # Block only high-severity
    MAX = "max"       # Block any unresolved conflict

def should_block(strictness: Strictness, conflicts: List[SemanticConflict]) -> bool:
    """Determine if generation should be blocked."""
    if strictness == Strictness.OFF:
        return False
    elif strictness == Strictness.MEDIUM:
        return any(c.severity == Severity.HIGH for c in conflicts)
    elif strictness == Strictness.MAX:
        return len(conflicts) > 0

Example usage:

generation_gate_middleware = GenerationGateMiddleware(
    strictness_policy=StrictnessPolicy(
        global_default=Strictness.MEDIUM,
        mission_override=None,
        step_override=None,
        runtime_override=None
    )
)

try:
    context = generation_gate_middleware.process(context)
    # Generation allowed - proceed to next step
except BlockedByConflict as e:
    # Generation blocked - proceed to clarification
    conflicts = e.conflicts

4. ClarificationMiddleware

Purpose: Render ranked candidates, prompt user for resolution (interactive or async).

Interface:

class ClarificationMiddleware:
    """Prompt user for conflict resolution."""

    def __init__(self, interaction_mode: InteractionMode):
        """
        Initialize clarification middleware.

        Args:
            interaction_mode: Interactive (CLI) or non-interactive (CI)
        """
        self.interaction_mode = interaction_mode

    def process(self, context: PrimitiveExecutionContext) -> PrimitiveExecutionContext:
        """
        Prompt user for conflict resolution.

        Requires:
            - context.conflicts (from generation gate)

        Emits events:
            - GlossaryClarificationRequested (on defer or non-interactive)
            - GlossaryClarificationResolved (on user selection)
            - GlossarySenseUpdated (on custom sense)

        Raises:
            DeferredToAsync: If user defers resolution

        Returns:
            Modified context with updated glossary (if resolved)
        """
        ...

Clarification flow:

from enum import Enum

class ClarificationChoice(Enum):
    CANDIDATE_SELECTED = "candidate"  # User picked from candidates
    CUSTOM_SENSE = "custom"           # User provided custom definition
    DEFER_ASYNC = "defer"             # User deferred to async resolution

def prompt_for_resolution(conflict: SemanticConflict) -> ClarificationChoice:
    """
    Render conflict and prompt for resolution.

    Displays:
        - Term, context, scope
        - Ranked candidate senses (by confidence)
        - Options: 1..N (candidates), C (custom), D (defer)

    Returns:
        User's choice
    """
    ...

Example usage:

clarification_middleware = ClarificationMiddleware(
    interaction_mode=InteractionMode.INTERACTIVE
)

try:
    context = clarification_middleware.process(context)
    # User resolved conflict - glossary updated
except DeferredToAsync:
    # User deferred resolution - exit with blocked status
    exit(1)

5. ResumeMiddleware

Purpose: Load checkpoint from events, restore step execution context.

Interface:

class ResumeMiddleware:
    """Resume execution from checkpoint after conflict resolution."""

    def __init__(self, checkpoint_store: CheckpointStore):
        """
        Initialize resume middleware.

        Args:
            checkpoint_store: Access to StepCheckpointed events
                - Loads from event log
                - Verifies input_hash integrity
        """
        self.checkpoint_store = checkpoint_store

    def process(self, context: PrimitiveExecutionContext) -> PrimitiveExecutionContext:
        """
        Resume from checkpoint.

        Requires:
            - context.retry_token (from user retry request)

        Modifies context:
            - Restores strictness, scope_refs from checkpoint
            - Loads updated glossary state from GlossarySenseUpdated events

        Emits events:
            - None (checkpoint already exists)

        Returns:
            Restored context (ready to resume from cursor)
        """
        ...

Resume flow:

def resume_from_checkpoint(retry_token: str, context: PrimitiveExecutionContext) -> PrimitiveExecutionContext:
    """
    Resume execution from checkpoint.

    Steps:
        1. Load StepCheckpointed event by retry_token
        2. Verify input_hash matches current inputs
        3. If changed: prompt user for confirmation
        4. Restore strictness, scope_refs from checkpoint
        5. Load updated glossary state from events
        6. Resume from cursor (skip already-completed stages)

    Returns:
        Restored context
    """
    ...

Example usage:

resume_middleware = ResumeMiddleware(
    checkpoint_store=CheckpointStore(event_log)
)

context = resume_middleware.process(context)
# context.strictness, context.scope_refs restored from checkpoint
# Skip to cursor: "pre_generation_gate"

Pipeline Composition

Middleware components are composed into a pipeline:

from typing import List

class GlossaryMiddlewarePipeline:
    """Compose glossary middleware components into a pipeline."""

    def __init__(self, middleware: List[GlossaryMiddleware]):
        """
        Initialize pipeline.

        Args:
            middleware: Ordered list of middleware components
        """
        self.middleware = middleware

    def process(self, context: PrimitiveExecutionContext) -> PrimitiveExecutionContext:
        """
        Execute middleware pipeline.

        Args:
            context: Initial execution context

        Returns:
            Final context (after all middleware)

        Raises:
            BlockedByConflict: If generation gate blocks
            DeferredToAsync: If clarification deferred
        """
        for mw in self.middleware:
            context = mw.process(context)
        return context

Example pipeline setup:

pipeline = GlossaryMiddlewarePipeline([
    GlossaryCandidateExtractionMiddleware(extraction_config),
    SemanticCheckMiddleware(glossary_store),
    GenerationGateMiddleware(strictness_policy),
    ClarificationMiddleware(interaction_mode),
    ResumeMiddleware(checkpoint_store)
])

try:
    final_context = pipeline.process(initial_context)
    # All checks passed - proceed to generation
except BlockedByConflict:
    # Generation blocked - clarification required
except DeferredToAsync:
    # User deferred resolution - exit

Exception Hierarchy

class GlossaryError(Exception):
    """Base exception for glossary errors."""

class BlockedByConflict(GlossaryError):
    """Generation blocked by unresolved high-severity conflict."""
    def __init__(self, conflicts: List[SemanticConflict]):
        self.conflicts = conflicts
        super().__init__(f"Generation blocked by {len(conflicts)} conflict(s)")

class DeferredToAsync(GlossaryError):
    """User deferred conflict resolution to async mode."""
    def __init__(self, conflict_id: str):
        self.conflict_id = conflict_id
        super().__init__(f"Conflict {conflict_id} deferred to async resolution")

class AbortResume(GlossaryError):
    """User aborted resume (context changed)."""
    def __init__(self, reason: str):
        self.reason = reason
        super().__init__(f"Resume aborted: {reason}")

Testing Middleware

Unit tests (per middleware component):

Mock inputs, assert outputs
Verify events emitted at correct stage
Test exception handling (BlockedByConflict, DeferredToAsync)

Integration tests (full pipeline):

Run extraction → check → gate → clarification → resume
Verify event order: TermCandidateObserved → SemanticCheckEvaluated → GenerationBlockedBySemanticConflict → GlossaryClarificationResolved
Test edge cases (no conflicts, all conflicts resolved, defer to async)

Example integration test:

def test_full_pipeline_with_conflict_resolution():
    # Setup
    context = PrimitiveExecutionContext(inputs={"description": "The workspace contains files"})
    pipeline = create_pipeline()

    # Execute
    try:
        final_context = pipeline.process(context)
        assert False, "Should have blocked"
    except BlockedByConflict:
        # Expected - clarification required
        pass

    # User resolves conflict
    resolve_conflict(conflict_id="uuid-1234", choice="candidate:1")

    # Retry with resume
    context.retry_token = "uuid-1234"
    final_context = pipeline.process(context)

    # Assert
    assert final_context.glossary_updated
    assert len(final_context.conflicts) == 0

Spec Kitty

Contracts

events.md

Event Contracts: Glossary Semantic Integrity

Overview

Canonical Events (Feature 007)

1. GlossaryScopeActivated

2. TermCandidateObserved

3. SemanticCheckEvaluated

4. GlossaryClarificationRequested

5. GlossaryClarificationResolved

6. GlossarySenseUpdated

7. GenerationBlockedBySemanticConflict

CLI-Specific Events (Pending Feature 007 Approval)

8. StepCheckpointed

Event Emission Guidelines

Testing Event Contracts

See Also

middleware.md

Middleware Contracts: Glossary Semantic Integrity

Overview

Base Middleware Protocol

Middleware Pipeline Order

1. GlossaryCandidateExtractionMiddleware

2. SemanticCheckMiddleware

3. GenerationGateMiddleware

4. ClarificationMiddleware

5. ResumeMiddleware

Pipeline Composition

Exception Hierarchy

Testing Middleware

See Also