AI-Driven Issue Tracking & Analytics Pipeline

Pass 1 — GPT-5 (Reasoning)
─────────────────────────────────────────────────────────
Input:   KSI control + Rev5 objective + client tech stack
Prompt:  "Analyze whether the client's infrastructure
          satisfies this control requirement. Explain
          the match type, coverage, and any gaps."
Output:  Prose analysis (saved to ai_response/output/)
Cost:    ~$0.03-0.05 per control part
Why GPT-5: Requires understanding compliance semantics,
           client infrastructure context, and gap
           identification — not a pattern-matching task.


Pass 2 — GPT-4.1 Mini (Extraction)
─────────────────────────────────────────────────────────
Input:   GPT-5's prose response from Pass 1
Prompt:  "Extract these fields from the analysis:
          match_type, coverage_status, gap_description,
          recommended_action. Return as CSV row."
Output:  Structured CSV (gaps_structured.csv)
Cost:    ~$0.001-0.003 per control part
Why 4.1 Mini: The answer is already in the text — this
              is field extraction, not reasoning. 10-30x
              cheaper and 3-5x faster than GPT-5.

# Verdict assignment logic (simplified from final_merge.ipynb)
# No AI involved — purely deterministic rules

def assign_verdict(row):
    ai_match = row['match_type']        # from Part 2 (GPT-5)
    vanta_status = row['vanta_status']   # from Part 3 (Vanta API)
    has_vanta = row['has_vanta_test']    # from Part 3 merge

    if ai_match == 'Direct' and vanta_status == 'PASSING':
        return 'COMPLIANT'

    if has_vanta and vanta_status == 'FAILING':
        return 'CRITICAL GAP (Operational Failure)'

    if ai_match == 'No Match' and not has_vanta:
        return 'STRATEGIC GAP (Missing Tool/Policy)'

    if ai_match == 'Direct' and not has_vanta:
        return 'LIKELY COMPLIANT (No Vanta Test)'

    return 'PARTIAL / VERIFICATION REQUIRED'

# Pattern used in every AI-calling notebook

# 1. Load existing results (skip completed work)
if os.path.exists(output_csv):
    done = pd.read_csv(output_csv)
    completed_ids = set(done['requirement_id'])
else:
    done = pd.DataFrame()
    completed_ids = set()

# 2. Filter to remaining work
remaining = df[~df['requirement_id'].isin(completed_ids)]
print(f"{len(completed_ids)} already done, "
      f"{len(remaining)} remaining")

# 3. Process with immediate persistence
def process_row(row):
    prompt = build_prompt(row)

    # Save prompt to disk BEFORE calling API
    save_prompt(prompt, row['requirement_id'])

    response = openai_client.chat(model="gpt-5", ...)

    # Save response to disk IMMEDIATELY after API returns
    save_output(response, row['requirement_id'])

    # Append to CSV immediately (not batched)
    append_to_csv(output_csv, parse_response(response))

    return response

# 4. Parallel execution with 5 workers
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {
        executor.submit(process_row, row): row
        for _, row in remaining.iterrows()
    }

# family-config cell (present in every notebook)
FAMILY = "KSI"  # or "ADS" or "CCM"

# All paths derived from FAMILY:
SOURCE = f"FILES/REFERENCE/fedramp/20x/TRUE_SOURCE_FILES/{FAMILY}/{FAMILY}_Golden_Template.csv"
OUTPUT = f"code/testing/ai_response/{FAMILY}/"
LABELS = [f"family:{FAMILY.lower()}"]  # family:ksi, family:ads, family:ccm

# Per-family output after full pipeline:
# ┌──────────┬──────────┬───────────────────────────┐
# │ Family   │ Controls │ Final Tickets             │
# ├──────────┼──────────┼───────────────────────────┤
# │ KSI      │ 56       │ 127 (13 Epic + 114 Task)  │
# │ ADS      │ 20       │ 31  (3 Epic + 7 Task +    │
# │          │          │      21 Sub-task)          │
# │ CCM      │ 3        │ 83  (3 Epic + 20 Task +   │
# │          │          │      60 Sub-task)          │
# └──────────┴──────────┴───────────────────────────┘

# Auto-detect subtask issue type name
# (Sunstone uses 'Subtask', SearchStax uses 'Sub-task')
issue_types = jira.get(f"/rest/api/3/issue/createmeta/...")
subtask_name = next(
    t['name'] for t in issue_types
    if t['name'].lower().replace('-', '') == 'subtask'
)

# Auto-detect project style
# (next-gen = team-managed, classic = company-managed)
project = jira.get(f"/rest/api/3/project/{PROJECT_KEY}")
project_style = project.get('style', 'classic')

# Auto-detect Epic Link custom field ID
# (varies per instance: customfield_10014, customfield_10600)
if project_style == 'classic':
    fields = jira.get("/rest/api/3/field")
    epic_link_field = next(
        f['id'] for f in fields
        if f['name'] == 'Epic Link'
    )

# Upload phases with proper hierarchy:
# Phase 1: Create Epics (root cause themes)
# Phase 2: Create Tasks linked to parent Epics
# Phase 3: Create Sub-tasks linked to parent Tasks