Collect a comprehensive git context snapshot. Mirrors TS getGitStatus() from context.ts. Runs multiple git commands in parallel for speed. Results are memoized; call clear_git_caches() to invalidate.
(
cwd: str | None = None,
)
| 122 | # --------------------------------------------------------------------------- |
| 123 | |
| 124 | async def collect_git_context( |
| 125 | cwd: str | None = None, |
| 126 | ) -> GitContextSnapshot: |
| 127 | """ |
| 128 | Collect a comprehensive git context snapshot. |
| 129 | |
| 130 | Mirrors TS getGitStatus() from context.ts. |
| 131 | Runs multiple git commands in parallel for speed. |
| 132 | Results are memoized; call clear_git_caches() to invalidate. |
| 133 | """ |
| 134 | global _git_context_cache |
| 135 | if _git_context_cache is not None: |
| 136 | return _git_context_cache |
| 137 | |
| 138 | target = cwd or os.getcwd() |
| 139 | |
| 140 | if not get_is_git(target): |
| 141 | snapshot = GitContextSnapshot(available=False, error="Not a git repository") |
| 142 | _git_context_cache = snapshot |
| 143 | return snapshot |
| 144 | |
| 145 | loop = asyncio.get_event_loop() |
| 146 | |
| 147 | branch_fut = loop.run_in_executor( |
| 148 | None, _git_cmd, ["rev-parse", "--abbrev-ref", "HEAD"], target, |
| 149 | ) |
| 150 | default_branch_fut = loop.run_in_executor( |
| 151 | None, _get_default_branch, target, |
| 152 | ) |
| 153 | status_fut = loop.run_in_executor( |
| 154 | # --no-optional-locks: don't take index write locks for the |
| 155 | # status probe — a concurrent `git` in the user's other terminal |
| 156 | # must never block on ours (TS context.ts:63-72; the flag rides |
| 157 | # only status + log there — TS-exact, ch03 round-3 G2). |
| 158 | None, _git_cmd, ["--no-optional-locks", "status", "--short"], target, |
| 159 | ) |
| 160 | commits_fut = loop.run_in_executor( |
| 161 | None, _git_cmd, ["--no-optional-locks", "log", "--oneline", "-n", "5"], target, |
| 162 | ) |
| 163 | user_fut = loop.run_in_executor( |
| 164 | None, _git_cmd, ["config", "user.name"], target, |
| 165 | ) |
| 166 | root_fut = loop.run_in_executor( |
| 167 | None, _git_cmd, ["rev-parse", "--show-toplevel"], target, |
| 168 | ) |
| 169 | |
| 170 | branch, default_branch, status, commits, user_name, repo_root = await asyncio.gather( |
| 171 | branch_fut, default_branch_fut, status_fut, commits_fut, user_fut, root_fut, |
| 172 | ) |
| 173 | |
| 174 | status_truncated = False |
| 175 | if status and len(status) > MAX_STATUS_CHARS: |
| 176 | status = status[:MAX_STATUS_CHARS] + "\n... (truncated)" |
| 177 | status_truncated = True |
| 178 | |
| 179 | snapshot = GitContextSnapshot( |
| 180 | available=True, |
| 181 | repo_root=repo_root or None, |