swadhin / contexa public

python: define core data models (OTARecord, CommitRecord, BranchMetadata)

Swadhin Biswas committed 7 months ago
3970c12
1 parent 419371c
commit 3970c12
1 file changed
+137 -0
Showing 1 changed file+137-0
PYTHON/contexa/models.py
▼PYTHON/contexa/models.py
+137-0
@@ -0,0 +1,137 @@
1
+"""
2
+GCC(Git-Context-Controller) Data Models Code by @swadhinbiswas
3
+
4
+Based on: arXiv:2508.00031v2 - "Git Context Controller: Manage the Context of
5
+LLM-based Agents like Git" by Junde Wu et al.
6
+
7
+The paper defines agent memory as a versioned, hierarchical file system with
8
+four core commands: COMMIT, BRANCH, MERGE, and CONTEXT.
9
+
10
+"""
11
+
12
+from dataclasses import dataclass, field
13
+from datetime import datetime
14
+from typing import Dict, Optional
15
+import yaml
16
+
17
+
18
+@dataclass
19
+class OTARecord:
20
+ """
21
+ Observation-Thought-Action record (OTA cycle).
22
+ The paper logs continuous OTA cycles in log.md for each branch.
23
+ """
24
+
25
+ timestamp: str
26
+ observation: str
27
+ thought: str
28
+ action: str
29
+ step: int
30
+
31
+ def to_markdown(self) -> str:
32
+ return (
33
+ f"### Step {self.step}-{self.timestamp}\n"
34
+ f"**Observation:** {self.observation}\n\n"
35
+ f"**Thought:** {self.thought}\n\n"
36
+ f"**Action:** {self.action}\n\n"
37
+ "--------\n"
38
+ )
39
+
40
+ @staticmethod
41
+ def from_dict(d: Dict) -> "OTARecord":
42
+ return OTARecord(
43
+ timestamp=d["timestamp"],
44
+ observation=d["observation"],
45
+ thought=d["thought"],
46
+ action=d["action"],
47
+ step=d["step"],
48
+ )
49
+
50
+
51
+@dataclass
52
+class CommitRecord:
53
+ commit_id: str
54
+ branch_name: str
55
+ branch_purpose: str
56
+ previous_progress_summary: str
57
+ this_commit_contribution: str
58
+ timestamp: str
59
+
60
+ def to_markdown(self) -> str:
61
+ return (
62
+ f"## Commit `{self.commit_id}`\n"
63
+ f"**Timestamp:** {self.timestamp}\n\n"
64
+ f"**Branch Purpose:** {self.branch_purpose}\n\n"
65
+ f"**Previous Progress Summary:** {self.previous_progress_summary}\n\n"
66
+ f"**This Commit's Contribution:** {self.this_commit_contribution}\n\n"
67
+ "---\n"
68
+ )
69
+
70
+
71
+@dataclass
72
+class BranchMetadata:
73
+ """
74
+ Structured metadata stored in metadata.yaml per branch.
75
+ Records architectural intent and motivation (per paper S3.1).
76
+ """
77
+
78
+ name: str
79
+ purpose: str
80
+ created_from: str
81
+ created_at: str
82
+ status: str = "active"
83
+ merged_into: Optional[str] = None
84
+ merged_at: Optional[str] = None
85
+
86
+ def to_yaml(self) -> str:
87
+ d = {
88
+ "name": self.name,
89
+ "purpose": self.purpose,
90
+ "created_from": self.created_from,
91
+ "created_at": self.created_at,
92
+ "status": self.status,
93
+ "merged_into": self.merged_into,
94
+ "merged_at": self.merged_at,
95
+ }
96
+ return yaml.dump(d, default_flow_style=False)
97
+
98
+ @staticmethod
99
+ def from_yaml(text: str) -> "BranchMetadata":
100
+ d = yaml.safe_load(text)
101
+ return BranchMetadata(
102
+ name=d["name"],
103
+ purpose=d["purpose"],
104
+ created_from=d["created_from"],
105
+ created_at=d["created_at"],
106
+ status=d.get("status", "active"),
107
+ merged_into=d.get("merged_into"),
108
+ merged_at=d.get("merged_at"),
109
+ )
110
+
111
+
112
+@dataclass
113
+class ContextResult:
114
+ """
115
+ Result returned by the CONTEXT command.
116
+ The paper specifies hierarchical retrieval at varying resolutions
117
+ (K = number of recent commit records to reveal; default K=1).
118
+ """
119
+
120
+ branch_name: str
121
+ k: int
122
+ commits: list[CommitRecord] = field(default_factory=list)
123
+ ota_records: list[OTARecord] = field(default_factory=list)
124
+ main_roadmap: str = ""
125
+ metadata: Optional[BranchMetadata] = None
126
+
127
+ def summary(self) -> str:
128
+ lines = [f"# CONTEXT — branch `{self.branch_name}` (K={self.k})\n"]
129
+ lines.append("## Global Roadmap\n" + self.main_roadmap + "\n")
130
+ lines.append(f"## Last {self.k} Commit(s)\n")
131
+ for c in self.commits:
132
+ lines.append(c.to_markdown())
133
+ if self.ota_records:
134
+ lines.append(f"## Recent OTA steps ({len(self.ota_records)} total)\n")
135
+ for rec in self.ota_records[-5:]:
136
+ lines.append(rec.to_markdown())
137
+ return "\n".join(lines)
0
138