Map Forgejo CI Status Into GitHub-Shaped Output

Forgejo commit statuses now feed the shim's GitHub-compatible PR status surfaces, so tools that ask for statusCheckRollup or gh pr checks receive useful check data instead of an empty placeholder.

ScopeForgejo client, PR routing, status normalization, tests, README.

Issuegh-forgejo-shim-wk9

Validation74 unit tests passed with unittest; compileall passed.

Summary

Added commit-status support for Forgejo pull requests and translated those statuses into GitHub CLI-shaped check rows and rollup items.

Changes Made

Added ForgejoClient.list_commit_statuses for /statuses/{sha} responses.
Added normalizers for statusCheckRollup objects and gh pr checks --json rows.
Updated PR list, view, status, and checks routing to fetch statuses from the PR head SHA when needed.
Updated tests to verify rollup output, check row output, and repeated-status de-duplication by context.
Updated the README to document status/check JSON behavior.

Context

GitHub-oriented tools often probe gh pr list, gh pr status, or gh pr checks to decide whether a branch is ready. Before this turn, the shim accepted those probes but did not translate Forgejo CI/status data into the fields those tools expect.

Important Implementation Details

Status data is fetched only when statusCheckRollup is requested or when gh pr checks runs, keeping ordinary PR calls cheap.
Forgejo states are mapped into stable buckets: success becomes pass, failures and errors become fail, and everything else becomes pending.
When Forgejo returns repeated status updates for a context, the latest timestamp wins.
If a pull request has no head SHA, the shim returns an empty status/check list instead of failing.

Relevant Diff Snippets

This is a focused server-rendered excerpt from the core implementation files.

src/gh_forgejo_shim/forgejo.py

63 unmodified lines
64
65
66
67
68
69
63 unmodified lines
            return []
        return [item for item in files if isinstance(item, dict)]

    def get_repo(self, repo: RepoRef) -> dict[str, Any]:
        return self._request_json("GET", repo.api_base_url, None)

63 unmodified lines
64
65
66
67
68
69
70
71
72
73
74
75
76
77
63 unmodified lines
            return []
        return [item for item in files if isinstance(item, dict)]

    def list_commit_statuses(self, repo: RepoRef, sha: str) -> list[dict[str, Any]]:
        statuses = self._request_json("GET", f"{repo.api_base_url}/statuses/{_quote(sha)}", None)
        if isinstance(statuses, dict) and isinstance(statuses.get("statuses"), list):
            statuses = statuses["statuses"]
        if not isinstance(statuses, list):
            return []
        return [item for item in statuses if isinstance(item, dict)]

    def get_repo(self, repo: RepoRef) -> dict[str, Any]:
        return self._request_json("GET", repo.api_base_url, None)

src/gh_forgejo_shim/normalize.py

-1+147

115 unmodified lines
116
117
118
119
120
121
21 unmodified lines
143
144
145
146
147
148
149
150
151
152
180 unmodified lines
333
334
335
336
337
338
115 unmodified lines
    "url",
)


def normalize_pull(pull: dict[str, Any]) -> dict[str, Any]:
    user = pull.get("user") if isinstance(pull.get("user"), dict) else {}
21 unmodified lines
        "mergeStateStatus": str(pull.get("merge_state_status") or "UNKNOWN").upper(),
        "reviewDecision": pull.get("reviewDecision"),
        "reviewRequests": pull.get("reviewRequests") or [],
        "statusCheckRollup": pull.get("statusCheckRollup") or [],
    }


def _normalize_count(value: Any) -> int:
    if isinstance(value, bool):
        return 0
180 unmodified lines
    }


def _is_organization(owner: dict[str, Any]) -> bool:
    owner_type = str(owner.get("type") or "").lower()
    return owner_type in {"organization", "org"}115 unmodified lines
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
21 unmodified lines
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
180 unmodified lines
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
115 unmodified lines
    "url",
)

SUPPORTED_CHECK_JSON_FIELDS = (
    "bucket",
    "completedAt",
    "conclusion",
    "description",
    "detailsUrl",
    "link",
    "name",
    "startedAt",
    "state",
    "workflow",
)


def normalize_pull(pull: dict[str, Any]) -> dict[str, Any]:
    user = pull.get("user") if isinstance(pull.get("user"), dict) else {}
21 unmodified lines
        "mergeStateStatus": str(pull.get("merge_state_status") or "UNKNOWN").upper(),
        "reviewDecision": pull.get("reviewDecision"),
        "reviewRequests": pull.get("reviewRequests") or [],
        "statusCheckRollup": _normalize_status_check_rollup(pull.get("statusCheckRollup")),
    }


def with_status_check_rollup(pull: dict[str, Any], statuses: list[dict[str, Any]]) -> dict[str, Any]:
    enriched = dict(pull)
    enriched["statusCheckRollup"] = normalize_status_check_rollup(statuses)
    return enriched


def normalize_status_check_rollup(statuses: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [_status_to_rollup_item(status) for status in _latest_statuses(statuses)]


def normalize_pr_checks(statuses: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [_status_to_check_item(status) for status in _latest_statuses(statuses)]


def filter_check_fields(data: dict[str, Any], fields: tuple[str, ...]) -> dict[str, Any]:
    if not fields:
        return data
    return {field: data.get(field) for field in fields if field in SUPPORTED_CHECK_JSON_FIELDS}


def _normalize_count(value: Any) -> int:
    if isinstance(value, bool):
        return 0
180 unmodified lines
    }


def _normalize_status_check_rollup(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        return []
    return [item for item in value if isinstance(item, dict)]


def _latest_statuses(statuses: list[dict[str, Any]]) -> list[dict[str, Any]]:
    latest: dict[str, dict[str, Any]] = {}
    for status in statuses:
        key = _status_name(status)
        existing = latest.get(key)
        if existing is None or _status_timestamp(status) >= _status_timestamp(existing):
            latest[key] = status
    return sorted(latest.values(), key=_status_name)


def _status_to_rollup_item(status: dict[str, Any]) -> dict[str, Any]:
    state = _status_state(status)
    target_url = _status_url(status)
    return {
        "__typename": "StatusContext",
        "completedAt": _status_completed_at(status, state),
        "conclusion": _status_conclusion(state),
        "context": _status_name(status),
        "description": status.get("description"),
        "detailsUrl": target_url,
        "name": _status_name(status),
        "startedAt": status.get("created_at"),
        "state": state,
        "targetUrl": target_url,
        "workflowName": _status_workflow(status),
    }


def _status_to_check_item(status: dict[str, Any]) -> dict[str, Any]:
    state = _status_state(status)
    target_url = _status_url(status)
    return {
        "bucket": _status_bucket(state),
        "completedAt": _status_completed_at(status, state),
        "conclusion": _status_conclusion(state),
        "description": status.get("description"),
        "detailsUrl": target_url,
        "link": target_url,
        "name": _status_name(status),
        "startedAt": status.get("created_at"),
        "state": _check_state(state),
        "workflow": _status_workflow(status),
    }


def _status_name(status: dict[str, Any]) -> str:
    context = status.get("context") or status.get("name") or status.get("title")
    return str(context) if context else "status"


def _status_workflow(status: dict[str, Any]) -> str:
    workflow = status.get("workflow") or status.get("workflow_name") or status.get("context")
    return str(workflow) if workflow else _status_name(status)


def _status_url(status: dict[str, Any]) -> str | None:
    url = status.get("target_url") or status.get("url") or status.get("html_url")
    return url if isinstance(url, str) and url else None


def _status_timestamp(status: dict[str, Any]) -> str:
    value = status.get("updated_at") or status.get("created_at")
    return value if isinstance(value, str) else ""


def _status_state(status: dict[str, Any]) -> str:
    raw = str(status.get("state") or status.get("status") or "").lower()
    if raw in {"success", "successful", "ok", "pass", "passed"}:
        return "SUCCESS"
    if raw in {"failure", "failed"}:
        return "FAILURE"
    if raw in {"error", "cancelled", "canceled", "warning"}:
        return "ERROR"
    return "PENDING"


def _status_conclusion(state: str) -> str | None:
    if state == "SUCCESS":
        return "success"
    if state == "FAILURE":
        return "failure"
    if state == "ERROR":
        return "failure"
    return None


def _status_bucket(state: str) -> str:
    if state == "SUCCESS":
        return "pass"
    if state in {"FAILURE", "ERROR"}:
        return "fail"
    return "pending"


def _check_state(state: str) -> str:
    if state in {"SUCCESS", "FAILURE", "ERROR"}:
        return "completed"
    return "pending"


def _status_completed_at(status: dict[str, Any], state: str) -> str | None:
    if state not in {"SUCCESS", "FAILURE", "ERROR"}:
        return None
    value = status.get("updated_at") or status.get("created_at")
    return value if isinstance(value, str) else None


def _is_organization(owner: dict[str, Any]) -> bool:
    owner_type = str(owner.get("type") or "").lower()
    return owner_type in {"organization", "org"}

src/gh_forgejo_shim/routing.py

-7+65

14 unmodified lines
15
16
17
18
19
20
21
22
23
24
25
26
27
125 unmodified lines
153
154
155
156
157
158
159
124 unmodified lines
284
285
286
287
288
289
290
2 unmodified lines
293
294
295
296
297
298
299
300
301
302
303
304
161 unmodified lines
466
467
468
469
470
471
472
53 unmodified lines
526
527
528
529
530
531
532
947 unmodified lines
1480
1481
1482
1483
1484
1485
143 unmodified lines
1629
1630
1631
1632
1633
1634
5 unmodified lines
1640
1641
1642
1643
1644
1645
14 unmodified lines
from .external import find_program, git_run, run_program
from .forgejo import ForgejoClient, ForgejoError, RepoRef
from .normalize import (
    filter_issue_fields,
    filter_fields,
    filter_repo_fields,
    normalize_issue,
    normalize_pull,
    normalize_repo,
    status_for_current_branch,
)
from .repo import (
    current_branch,
125 unmodified lines
    if command in {"create", "new"}:
        return _run_create(rest, repo, client, cwd=cwd, stdout=stdout, stdin=stdin)
    if command == "checks":
        return _run_checks(rest, stdout=stdout)
    if command in {"checkout", "co"}:
        return _run_checkout(rest, repo, client, cwd=cwd, stdout=stdout, stderr=stderr)
    if command == "comment":
124 unmodified lines
    if parsed.limit is not None:
        pulls = pulls[: parsed.limit]

    normalized = [filter_fields(normalize_pull(pull), parsed.json_fields) for pull in pulls]
    if parsed.json_fields:
        _print_json_or_jq_list(normalized, parsed.jq, stdout)
    else:
2 unmodified lines
    return 0


def _run_checks(argv: list[str], *, stdout: TextIO) -> int:
    parsed = _parse_checks_args(argv)
    if parsed.json_fields:
        _print_json_or_jq_list([], parsed.jq, stdout)
    else:
        print("no checks reported", file=stdout)
    return 0


161 unmodified lines
            _print_json_or_jq({}, parsed.jq, stdout)
        return 0

    normalized = normalize_pull(pull)
    if parsed.web:
        url = str(normalized.get("url") or "")
        if url:
53 unmodified lines
        return 0

    if parsed.json_fields:
        _print_json_or_jq(status_for_current_branch(pull, parsed.json_fields), parsed.jq, stdout)
    else:
        normalized = normalize_pull(pull)
        print(_format_pull_text(normalized), file=stdout)
947 unmodified lines
    return f"{number}\t{title}\t{state}".rstrip()


def _print_json_or_jq(data: dict[str, object], jq: str | None, stdout: TextIO) -> None:
    if jq:
        value = _apply_simple_jq(data, jq)
143 unmodified lines
    return pulls[0] if pulls else None


def _pull_number(pull: dict[str, object]) -> int | None:
    raw = pull.get("number") or pull.get("id")
    if isinstance(raw, bool):
5 unmodified lines
    return None


def _pull_head_ref(pull: dict[str, object]) -> str | None:
    head = pull.get("head")
    if not isinstance(head, dict):14 unmodified lines
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
125 unmodified lines
156
157
158
159
160
161
162
124 unmodified lines
287
288
289
290
291
292
293
294
295
296
2 unmodified lines
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
161 unmodified lines
489
490
491
492
493
494
495
53 unmodified lines
549
550
551
552
553
554
555
556
947 unmodified lines
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
143 unmodified lines
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
5 unmodified lines
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
14 unmodified lines
from .external import find_program, git_run, run_program
from .forgejo import ForgejoClient, ForgejoError, RepoRef
from .normalize import (
    filter_check_fields,
    filter_issue_fields,
    filter_fields,
    filter_repo_fields,
    normalize_pr_checks,
    normalize_issue,
    normalize_pull,
    normalize_repo,
    status_for_current_branch,
    with_status_check_rollup,
)
from .repo import (
    current_branch,
125 unmodified lines
    if command in {"create", "new"}:
        return _run_create(rest, repo, client, cwd=cwd, stdout=stdout, stdin=stdin)
    if command == "checks":
        return _run_checks(rest, repo, client, cwd=cwd, stdout=stdout)
    if command in {"checkout", "co"}:
        return _run_checkout(rest, repo, client, cwd=cwd, stdout=stdout, stderr=stderr)
    if command == "comment":
124 unmodified lines
    if parsed.limit is not None:
        pulls = pulls[: parsed.limit]

    normalized = [
        filter_fields(normalize_pull(_enrich_pull_statuses(target_repo, client, pull, parsed.json_fields)), parsed.json_fields)
        for pull in pulls
    ]
    if parsed.json_fields:
        _print_json_or_jq_list(normalized, parsed.jq, stdout)
    else:
2 unmodified lines
    return 0


def _run_checks(
    argv: list[str],
    repo: RepoRef,
    client: ForgejoClient,
    *,
    cwd: str | None,
    stdout: TextIO,
) -> int:
    parsed = _parse_checks_args(argv)
    target_repo = parse_repo_spec(parsed.repo, default_host=repo.host) if parsed.repo else repo
    if target_repo is None:
        raise ValueError("could not parse --repo value")

    pull = _resolve_pull(target_repo, client, number=parsed.number, branch=None, cwd=cwd)
    statuses = [] if pull is None else _pull_statuses(target_repo, client, pull)
    checks = [filter_check_fields(check, parsed.json_fields) for check in normalize_pr_checks(statuses)]
    if parsed.json_fields:
        _print_json_or_jq_list(checks, parsed.jq, stdout)
    else:
        if not checks:
            print("no checks reported", file=stdout)
        for check in checks:
            print(_format_check_item(check), file=stdout)
    return 0


161 unmodified lines
            _print_json_or_jq({}, parsed.jq, stdout)
        return 0

    normalized = normalize_pull(_enrich_pull_statuses(repo, client, pull, parsed.json_fields))
    if parsed.web:
        url = str(normalized.get("url") or "")
        if url:
53 unmodified lines
        return 0

    if parsed.json_fields:
        enriched = _enrich_pull_statuses(repo, client, pull, parsed.json_fields)
        _print_json_or_jq(status_for_current_branch(enriched, parsed.json_fields), parsed.jq, stdout)
    else:
        normalized = normalize_pull(pull)
        print(_format_pull_text(normalized), file=stdout)
947 unmodified lines
    return f"{number}\t{title}\t{state}".rstrip()


def _format_check_item(data: dict[str, object]) -> str:
    bucket = data.get("bucket") or "pending"
    name = data.get("name") or "status"
    description = data.get("description") or ""
    return f"{bucket}\t{name}\t{description}".rstrip()


def _print_json_or_jq(data: dict[str, object], jq: str | None, stdout: TextIO) -> None:
    if jq:
        value = _apply_simple_jq(data, jq)
143 unmodified lines
    return pulls[0] if pulls else None


def _enrich_pull_statuses(
    repo: RepoRef,
    client: ForgejoClient,
    pull: dict[str, object],
    fields: tuple[str, ...],
) -> dict[str, object]:
    if "statusCheckRollup" not in fields:
        return pull
    statuses = _pull_statuses(repo, client, pull)
    return with_status_check_rollup(pull, statuses)


def _pull_statuses(repo: RepoRef, client: ForgejoClient, pull: dict[str, object]) -> list[dict[str, object]]:
    sha = _pull_head_sha(pull)
    if sha is None:
        return []
    return client.list_commit_statuses(repo, sha)


def _pull_number(pull: dict[str, object]) -> int | None:
    raw = pull.get("number") or pull.get("id")
    if isinstance(raw, bool):
5 unmodified lines
    return None


def _pull_head_sha(pull: dict[str, object]) -> str | None:
    head = pull.get("head")
    if not isinstance(head, dict):
        return None
    sha = head.get("sha")
    return sha if isinstance(sha, str) and sha else None


def _pull_head_ref(pull: dict[str, object]) -> str | None:
    head = pull.get("head")
    if not isinstance(head, dict):

Expected Impact for End-Users

Codex, T3, editors, and bots that expect GitHub-style PR checks should now see meaningful Forgejo status data for allowlisted repositories, including pass, fail, and pending check rows.

Validation

PYTHONPATH=src python3 -m unittest discover -s tests: 74 tests passed.
python3 -m compileall -q src tests: passed.
python3 -m pytest -q: not available because the active Python environment has no pytest module.

Issues, Limitations, and Mitigations

bd dolt pull failed because local and remote Dolt histories have no common ancestor. I left the local Beads issue in place and did not run destructive recovery.
The mapping uses Forgejo commit statuses. If a Forgejo instance exposes richer Actions jobs separately without commit statuses, those jobs will not appear until the instance publishes statuses or the shim adds a second API source.
The rollup shape is intentionally practical rather than a complete GitHub GraphQL object model.

Follow-up Work

Resolve Beads Dolt divergence before relying on cross-machine issue sync.
Smoke test against a live Forgejo repository with passing, failing, and pending statuses.
Add a compatibility note if a specific editor expects additional gh pr checks fields.