-
Notifications
You must be signed in to change notification settings - Fork 18
Expand file tree
/
Copy pathcontext.py
More file actions
79 lines (65 loc) · 2.59 KB
/
context.py
File metadata and controls
79 lines (65 loc) · 2.59 KB
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
"""Context hydration: GitHub issue fetching and prompt assembly."""
import requests
from models import GitHubIssue, IssueComment, TaskConfig
def fetch_github_issue(repo_url: str, issue_number: str, token: str) -> GitHubIssue:
"""Fetch a GitHub issue's title, body, and comments."""
headers = {
"Authorization": f"token {token}",
"Accept": "application/vnd.github.v3+json",
}
# Fetch issue
issue_resp = requests.get(
f"https://api.github.com/repos/{repo_url}/issues/{issue_number}",
headers=headers,
timeout=30,
)
issue_resp.raise_for_status()
issue = issue_resp.json()
# Fetch comments
comments: list[IssueComment] = []
if issue.get("comments", 0) > 0:
comments_resp = requests.get(
f"https://api.github.com/repos/{repo_url}/issues/{issue_number}/comments",
headers=headers,
timeout=30,
)
comments_resp.raise_for_status()
comments = [
IssueComment(id=int(c["id"]), author=c["user"]["login"], body=c["body"] or "")
for c in comments_resp.json()
]
return GitHubIssue(
title=issue["title"],
body=issue.get("body", "") or "",
number=issue["number"],
comments=comments,
)
def assemble_prompt(config: TaskConfig) -> str:
"""Assemble the user prompt from issue context and task description.
.. deprecated::
In production (AgentCore server mode), the orchestrator's
``assembleUserPrompt()`` in ``context-hydration.ts`` is the sole prompt
assembler. The hydrated prompt arrives via
``HydratedContext.user_prompt`` (validated from the incoming JSON).
This Python implementation is retained only for **local batch mode**
(``python src/entrypoint.py``) and **dry-run mode** (``DRY_RUN=1``).
"""
parts = []
parts.append(f"Task ID: {config.task_id}")
parts.append(f"Repository: {config.repo_url}")
if config.issue:
issue = config.issue
parts.append(f"\n## GitHub Issue #{issue.number}: {issue.title}\n")
parts.append(issue.body or "(no description)")
if issue.comments:
parts.append("\n### Comments\n")
for c in issue.comments:
parts.append(f"**@{c.author}**: {c.body}\n")
if config.task_description:
parts.append(f"\n## Task\n\n{config.task_description}")
elif config.issue:
parts.append(
"\n## Task\n\nResolve the GitHub issue described above. "
"Follow the workflow in your system instructions."
)
return "\n".join(parts)