Skip to content

feat: emit structured Policy ACK telemetry #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zehongye-gaga merged 1 commit into from
Nov 12, 2026
Merged

feat: emit structured Policy ACK telemetry #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 79 additions & 15 deletions .github/workflows/codex-code-review.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ jobs:
python3 - <<'OCR_PY'
# SPDX-License-Identifier: Apache-2.0
"""Trusted workflow client; never check out or execute pull request code."""
from datetime import datetime, timezone
import hashlib
import http.client
import json
Expand All @@ -47,7 +48,42 @@ jobs:


class ReviewError(Exception):
pass
def __init__(self, message, code='invalid_response'):
super().__init__(message)
self.code = code


_telemetry = None


class Telemetry:
def __init__(self):
self.started = time.monotonic()
self.sequence = 0
self.finished = False
self.context = {key: os.environ.get(env) for key, env in [
('repository', 'GITHUB_REPOSITORY'), ('repository_id', 'GITHUB_REPOSITORY_ID'),
('run_id', 'GITHUB_RUN_ID'), ('run_attempt', 'GITHUB_RUN_ATTEMPT'),
('job', 'GITHUB_JOB'), ('workflow_ref', 'GITHUB_WORKFLOW_REF')]}

def emit(self, event, **fields):
self.sequence += 1
record = {**self.context, **fields, 'schema': 'ocr-policy-telemetry/v1',
'event': event, 'sequence': self.sequence,
'at': datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z')}
try:
print(json.dumps(record, sort_keys=True), flush=True)
except OSError:
# Observability must not change the admission result.
pass

def finish(self, outcome, reason=None):
if self.finished:
return
self.finished = True
self.emit('ocr_client_result', outcome=outcome, reason=reason,
elapsed_ms=round((time.monotonic() - self.started) * 1000))



class NoRedirect(urllib.request.HTTPRedirectHandler):
Expand Down Expand Up @@ -119,15 +155,19 @@ jobs:


def timeout_handler(signum, frame):
raise ReviewError('OCR acceptance exceeded 90 seconds; re-run to reuse the task.')
raise ReviewError('OCR acceptance exceeded 90 seconds; re-run to reuse the task.', 'ack_deadline_exceeded')


def log_request(path, payload, attempt, started, status, error=None, retry_delay=None):
def log_request(path, payload, attempt, started, status, error=None, retry_delay=None, error_code=None):
record = {'event': 'ocr_gateway_request', 'operation': path,
'request_id': payload.get('request_id'), 'attempt': attempt,
'elapsed_ms': round((time.monotonic() - started) * 1000),
'http_status': status, 'error': error, 'retry_delay_seconds': retry_delay}
print(json.dumps(record, sort_keys=True), flush=True)
record['error_code'] = error_code
if _telemetry:
_telemetry.emit(**record)
else:
print(json.dumps(record, sort_keys=True), flush=True)


def post(path, payload, token, deadline):
Expand All @@ -136,7 +176,7 @@ jobs:
started = time.monotonic()
remaining = deadline - started
if remaining <= 0:
raise ReviewError('Timed out waiting for OCR. Re-run this workflow to reuse the task.')
raise ReviewError('Timed out waiting for OCR. Re-run this workflow to reuse the task.', 'ack_deadline_exceeded')
status = None
req = urllib.request.Request(BASE_URL + path, data=data, headers={
'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'})
Expand All @@ -162,15 +202,17 @@ jobs:
status = error.code
error.close()
if status not in [408, 429, 500, 502, 503, 504]:
log_request(path, payload, attempt + 1, started, status, 'HTTPError')
raise ReviewError(f'OCR gateway HTTP {status}; check configuration and credentials.') from None
failure = 'HTTPError'
log_request(path, payload, attempt + 1, started, status, 'HTTPError', error_code='http_error')
raise ReviewError(f'OCR gateway HTTP {status}; check configuration and credentials.', 'http_error') from None
failure, failure_code = 'HTTPError', 'http_error'
except (urllib.error.URLError, ConnectionError, TimeoutError,
ssl.SSLEOFError, http.client.IncompleteRead, http.client.BadStatusLine,
json.JSONDecodeError, UnicodeError) as error:
failure = type(error).__name__
except ReviewError:
log_request(path, payload, attempt + 1, started, status, 'ReviewError')
cause = error.reason if isinstance(error, urllib.error.URLError) else error
failure_code = 'http_timeout' if isinstance(cause, TimeoutError) else 'transport_error'
except ReviewError as error:
log_request(path, payload, attempt + 1, started, status, 'ReviewError', error_code=error.code)
raise
else:
log_request(path, payload, attempt + 1, started, status)
Expand All @@ -180,11 +222,11 @@ jobs:
if attempt + 1 < MAX_ATTEMPTS and remaining > 0:
backoff = 2 ** attempt
delay = min(random.uniform(backoff / 2, backoff), remaining)
log_request(path, payload, attempt + 1, started, status, failure, delay)
log_request(path, payload, attempt + 1, started, status, failure, delay, failure_code)
if delay is None:
break
time.sleep(delay)
raise ReviewError('OCR gateway remains unavailable after bounded retries; re-run this workflow.')
raise ReviewError('OCR gateway remains unavailable after bounded retries; re-run this workflow.', 'gateway_retries_exhausted')


def summary(identity, reply):
Expand All @@ -203,13 +245,16 @@ jobs:
output.write(text)


def main():
def submit():
token = os.environ.get('OCR_CI_TOKEN', '').strip()
if not token:
raise ReviewError('Missing organization secret OCR_CI_TOKEN or repository access to it.')
raise ReviewError('Missing organization secret OCR_CI_TOKEN or repository access to it.', 'configuration_error')
with open(os.environ['GITHUB_EVENT_PATH']) as source:
identity = coordinates(json.load(source), os.environ)
receipt = {'schema': SCHEMA, 'request_id': request_id(identity), **identity}
if _telemetry:
_telemetry.context.update({**identity, 'request_id': receipt['request_id']})
_telemetry.emit('ocr_client_started')
payload = {**receipt, 'source': {'run_id': os.environ.get('GITHUB_RUN_ID'),
'run_attempt': os.environ.get('GITHUB_RUN_ATTEMPT'),
'workflow_ref': os.environ.get('GITHUB_WORKFLOW_REF')}}
Expand All @@ -227,6 +272,9 @@ jobs:
print('OCR dispatch state: ' + state, flush=True)
last_state = state
if result is not None:
skipped = state == 'skipped' and reply.get('reason') == 'repository_not_enabled'
if _telemetry:
_telemetry.finish('skipped' if skipped else 'accepted', 'repository_not_enabled' if skipped else None)
summary(identity, reply)
if state == 'skipped' and reply.get('reason') == 'repository_not_enabled':
print('OCR dispatch skipped: repository_not_enabled.')
Expand All @@ -235,11 +283,27 @@ jobs:
return result
remaining = deadline - time.monotonic()
if remaining <= 0:
raise ReviewError('OCR acceptance exceeded 90 seconds. Re-run to reuse the task.')
raise ReviewError('OCR acceptance exceeded 90 seconds. Re-run to reuse the task.', 'ack_deadline_exceeded')
time.sleep(min(5, remaining))
reply = post('ocr-ci-status', receipt, token, deadline)


def main():
global _telemetry
previous = _telemetry
_telemetry = Telemetry()
try:
return submit()
except ReviewError as error:
_telemetry.finish('timeout' if error.code == 'ack_deadline_exceeded' else 'failed', error.code)
raise
except Exception:
_telemetry.finish('failed', 'client_error')
raise
finally:
_telemetry = previous


if __name__ == '__main__':
try:
# Ubuntu runner: enforce total wall time even on trickling HTTP bodies.
Expand Down
4 changes: 4 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,3 +39,7 @@ An administrator can set the new `OCR public repositories` Ruleset to **Disabled
## Backend contract v2

Durable acceptance confirms backend scheduling responsibility, independently of whether the review is waiting, completed, skipped, blocked or failed. Legacy successful skipped responses are also supported. The workflow does not enforce the review verdict; configure the separate `ocr-review` required status when desired. Explicit Actions reruns provide a stable run/attempt retry key so failed pre-dispatch receipts can recover without duplicate submissions. The backend does not depend on n8n polling heartbeats.

## Audit telemetry

Structured results are collected asynchronously from Actions logs; the client makes no audit reporting request. See [the telemetry contract](ocr/telemetry.md).
94 changes: 79 additions & 15 deletions ocr/ci_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
"""Trusted workflow client; never check out or execute pull request code."""
from datetime import datetime, timezone
import hashlib
import http.client
import json
Expand All @@ -23,7 +24,42 @@


class ReviewError(Exception):
pass
def __init__(self, message, code='invalid_response'):
super().__init__(message)
self.code = code


_telemetry = None


class Telemetry:
def __init__(self):
self.started = time.monotonic()
self.sequence = 0
self.finished = False
self.context = {key: os.environ.get(env) for key, env in [
('repository', 'GITHUB_REPOSITORY'), ('repository_id', 'GITHUB_REPOSITORY_ID'),
('run_id', 'GITHUB_RUN_ID'), ('run_attempt', 'GITHUB_RUN_ATTEMPT'),
('job', 'GITHUB_JOB'), ('workflow_ref', 'GITHUB_WORKFLOW_REF')]}

def emit(self, event, **fields):
self.sequence += 1
record = {**self.context, **fields, 'schema': 'ocr-policy-telemetry/v1',
'event': event, 'sequence': self.sequence,
'at': datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z')}
try:
print(json.dumps(record, sort_keys=True), flush=True)
except OSError:
# Observability must not change the admission result.
pass

def finish(self, outcome, reason=None):
if self.finished:
return
self.finished = True
self.emit('ocr_client_result', outcome=outcome, reason=reason,
elapsed_ms=round((time.monotonic() - self.started) * 1000))



class NoRedirect(urllib.request.HTTPRedirectHandler):
Expand Down Expand Up @@ -95,15 +131,19 @@ def verdict(reply):


def timeout_handler(signum, frame):
raise ReviewError('OCR acceptance exceeded 90 seconds; re-run to reuse the task.')
raise ReviewError('OCR acceptance exceeded 90 seconds; re-run to reuse the task.', 'ack_deadline_exceeded')


def log_request(path, payload, attempt, started, status, error=None, retry_delay=None):
def log_request(path, payload, attempt, started, status, error=None, retry_delay=None, error_code=None):
record = {'event': 'ocr_gateway_request', 'operation': path,
'request_id': payload.get('request_id'), 'attempt': attempt,
'elapsed_ms': round((time.monotonic() - started) * 1000),
'http_status': status, 'error': error, 'retry_delay_seconds': retry_delay}
print(json.dumps(record, sort_keys=True), flush=True)
record['error_code'] = error_code
if _telemetry:
_telemetry.emit(**record)
else:
print(json.dumps(record, sort_keys=True), flush=True)


def post(path, payload, token, deadline):
Expand All @@ -112,7 +152,7 @@ def post(path, payload, token, deadline):
started = time.monotonic()
remaining = deadline - started
if remaining <= 0:
raise ReviewError('Timed out waiting for OCR. Re-run this workflow to reuse the task.')
raise ReviewError('Timed out waiting for OCR. Re-run this workflow to reuse the task.', 'ack_deadline_exceeded')
status = None
req = urllib.request.Request(BASE_URL + path, data=data, headers={
'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'})
Expand All @@ -138,15 +178,17 @@ def post(path, payload, token, deadline):
status = error.code
error.close()
if status not in [408, 429, 500, 502, 503, 504]:
log_request(path, payload, attempt + 1, started, status, 'HTTPError')
raise ReviewError(f'OCR gateway HTTP {status}; check configuration and credentials.') from None
failure = 'HTTPError'
log_request(path, payload, attempt + 1, started, status, 'HTTPError', error_code='http_error')
raise ReviewError(f'OCR gateway HTTP {status}; check configuration and credentials.', 'http_error') from None
failure, failure_code = 'HTTPError', 'http_error'
except (urllib.error.URLError, ConnectionError, TimeoutError,
ssl.SSLEOFError, http.client.IncompleteRead, http.client.BadStatusLine,
json.JSONDecodeError, UnicodeError) as error:
failure = type(error).__name__
except ReviewError:
log_request(path, payload, attempt + 1, started, status, 'ReviewError')
cause = error.reason if isinstance(error, urllib.error.URLError) else error
failure_code = 'http_timeout' if isinstance(cause, TimeoutError) else 'transport_error'
except ReviewError as error:
log_request(path, payload, attempt + 1, started, status, 'ReviewError', error_code=error.code)
raise
else:
log_request(path, payload, attempt + 1, started, status)
Expand All @@ -156,11 +198,11 @@ def post(path, payload, token, deadline):
if attempt + 1 < MAX_ATTEMPTS and remaining > 0:
backoff = 2 ** attempt
delay = min(random.uniform(backoff / 2, backoff), remaining)
log_request(path, payload, attempt + 1, started, status, failure, delay)
log_request(path, payload, attempt + 1, started, status, failure, delay, failure_code)
if delay is None:
break
time.sleep(delay)
raise ReviewError('OCR gateway remains unavailable after bounded retries; re-run this workflow.')
raise ReviewError('OCR gateway remains unavailable after bounded retries; re-run this workflow.', 'gateway_retries_exhausted')


def summary(identity, reply):
Expand All @@ -179,13 +221,16 @@ def summary(identity, reply):
output.write(text)


def main():
def submit():
token = os.environ.get('OCR_CI_TOKEN', '').strip()
if not token:
raise ReviewError('Missing organization secret OCR_CI_TOKEN or repository access to it.')
raise ReviewError('Missing organization secret OCR_CI_TOKEN or repository access to it.', 'configuration_error')
with open(os.environ['GITHUB_EVENT_PATH']) as source:
identity = coordinates(json.load(source), os.environ)
receipt = {'schema': SCHEMA, 'request_id': request_id(identity), **identity}
if _telemetry:
_telemetry.context.update({**identity, 'request_id': receipt['request_id']})
_telemetry.emit('ocr_client_started')
payload = {**receipt, 'source': {'run_id': os.environ.get('GITHUB_RUN_ID'),
'run_attempt': os.environ.get('GITHUB_RUN_ATTEMPT'),
'workflow_ref': os.environ.get('GITHUB_WORKFLOW_REF')}}
Expand All @@ -203,6 +248,9 @@ def main():
print('OCR dispatch state: ' + state, flush=True)
last_state = state
if result is not None:
skipped = state == 'skipped' and reply.get('reason') == 'repository_not_enabled'
if _telemetry:
_telemetry.finish('skipped' if skipped else 'accepted', 'repository_not_enabled' if skipped else None)
summary(identity, reply)
if state == 'skipped' and reply.get('reason') == 'repository_not_enabled':
print('OCR dispatch skipped: repository_not_enabled.')
Expand All @@ -211,11 +259,27 @@ def main():
return result
remaining = deadline - time.monotonic()
if remaining <= 0:
raise ReviewError('OCR acceptance exceeded 90 seconds. Re-run to reuse the task.')
raise ReviewError('OCR acceptance exceeded 90 seconds. Re-run to reuse the task.', 'ack_deadline_exceeded')
time.sleep(min(5, remaining))
reply = post('ocr-ci-status', receipt, token, deadline)


def main():
global _telemetry
previous = _telemetry
_telemetry = Telemetry()
try:
return submit()
except ReviewError as error:
_telemetry.finish('timeout' if error.code == 'ack_deadline_exceeded' else 'failed', error.code)
raise
except Exception:
_telemetry.finish('failed', 'client_error')
raise
finally:
_telemetry = previous


if __name__ == '__main__':
try:
# Ubuntu runner: enforce total wall time even on trickling HTTP bodies.
Expand Down
Loading
Loading