File size: 4,980 Bytes
c29f1fd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
"""Curated prototypes and example inputs for TorchReview Copilot."""

from __future__ import annotations

from typing import Dict, List

try:
    from .triage_models import IssueLabel, TriageExample, TriagePrototype
    from .tasks import list_tasks
except ImportError:
    from triage_models import IssueLabel, TriageExample, TriagePrototype
    from tasks import list_tasks


TASK_KIND_TO_LABEL: Dict[str, IssueLabel] = {
    "syntax_fix": "syntax",
    "bug_fix": "logic",
    "optimization": "performance",
}

TRACEBACK_BY_TASK_ID: Dict[str, str] = {
    "syntax_fix_invoice_totals": (
        "Traceback (most recent call last):\n"
        "  File \"services/billing/reconciliation.py\", line 3\n"
        "    for record in records\n"
        "                      ^\n"
        "SyntaxError: expected ':'"
    ),
    "bug_fix_session_windows": (
        "AssertionError: collapse_sessions([{'minute': 1}, {'minute': 3}, {'minute': 8}], 4)\n"
        "Expected: [(1, 3), (8, 8)]\n"
        "Actual:   [(1, 8)]\n"
        "Boundary handling merges the final session instead of starting a new one."
    ),
    "optimization_rank_active_users": (
        "BenchmarkWarning: rank_active_users exceeded the 450ms budget on a nightly export fixture.\n"
        "Profiler hint: repeated scans over the full event list and nested loops dominate runtime."
    ),
}

SUMMARY_BY_TASK_ID: Dict[str, str] = {
    "syntax_fix_invoice_totals": "Broken parser state in a billing helper blocks reconciliation jobs.",
    "bug_fix_session_windows": "Session-boundary logic fails on inclusive idle-timeout edges.",
    "optimization_rank_active_users": "A nightly ranking job is correct on small fixtures but too slow at production scale.",
}

CONTEXT_BY_TASK_ID: Dict[str, str] = {
    "syntax_fix_invoice_totals": (
        "Context window: this helper runs in an end-of-day billing reconciliation job. "
        "Keep the public function signature intact and restore correct totals for mixed integer/string inputs."
    ),
    "bug_fix_session_windows": (
        "Context window: this function groups sorted product analytics events into sessions for retention dashboards. "
        "Boundary behavior must stay deterministic because downstream reports depend on it."
    ),
    "optimization_rank_active_users": (
        "Context window: this pipeline feeds a nightly export on a small CPU instance. "
        "Maintain identical output ordering while improving scalability on larger event volumes."
    ),
}


def _prototype_text(

    task_id: str,

    title: str,

    description: str,

    repo_summary: str,

    goal: str,

    visible_tests: List[str],

    starter_code: str,

    traceback_text: str,

) -> str:
    visible = "\n".join(f"- {item}" for item in visible_tests) or "- none"
    return (
        f"Title: {title}\n"
        f"Problem: {description}\n"
        f"Repo context: {repo_summary}\n"
        f"Goal: {goal}\n"
        f"Observed failure:\n{traceback_text}\n"
        f"Visible checks:\n{visible}\n"
        f"Candidate code:\n{starter_code}\n"
        f"Task id: {task_id}\n"
    )


def build_examples() -> List[TriageExample]:
    """Create stable UI examples from the task catalog."""

    examples: List[TriageExample] = []
    for task in list_tasks():
        label = TASK_KIND_TO_LABEL[task.task_kind]
        examples.append(
            TriageExample(
                key=task.task_id,
                title=task.title,
                label=label,
                summary=SUMMARY_BY_TASK_ID[task.task_id],
                code=task.starter_code,
                traceback_text=TRACEBACK_BY_TASK_ID[task.task_id],
                context_window=CONTEXT_BY_TASK_ID[task.task_id],
                task_id=task.task_id,
            )
        )
    return examples


def build_prototypes() -> List[TriagePrototype]:
    """Build canonical triage prototypes from the OpenEnv tasks."""

    prototypes: List[TriagePrototype] = []
    for task in list_tasks():
        traceback_text = TRACEBACK_BY_TASK_ID[task.task_id]
        prototypes.append(
            TriagePrototype(
                task_id=task.task_id,
                title=task.title,
                label=TASK_KIND_TO_LABEL[task.task_kind],
                summary=SUMMARY_BY_TASK_ID[task.task_id],
                reference_text=_prototype_text(
                    task.task_id,
                    task.title,
                    task.task_description,
                    task.repo_summary,
                    task.goal,
                    list(task.visible_tests),
                    task.reference_code,
                    traceback_text,
                ),
                starter_code=task.starter_code,
                reference_code=task.reference_code,
                traceback_text=traceback_text,
            )
        )
    return prototypes