Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
23 changes: 23 additions & 0 deletions .github/workflows/biggiepockets-review.yml
Original file line number Diff line number Diff line change
Expand Up @@ -339,6 +339,29 @@ jobs:
ANTHROPIC_BASE_URL: "https://openrouter.ai/api"
ANTHROPIC_AUTH_TOKEN: ${{ secrets.OPENROUTER_API_KEY }}

- name: Capture review failure diagnostics
if: ${{ always() && steps.claude.outcome == 'failure' }}
env:
EXECUTION_FILE: ${{ steps.claude.outputs.execution_file }}
DIAGNOSTIC_TOKEN_PROVIDER: ${{ secrets.OPENROUTER_API_KEY }}
DIAGNOSTIC_TOKEN_APP: ${{ steps.claude.outputs.github_token }}
DIAGNOSTIC_TOKEN_GITHUB: ${{ github.token }}
DIAGNOSTIC_TOKEN_REVIEW: ${{ secrets.BIGGIEPOCKETS_PAT }}
run: |
# The action may fail before exposing its execution_file output.
execution_file="${EXECUTION_FILE:-$RUNNER_TEMP/claude-execution-output.json}"
python3 registry/scripts/review-diagnostics.py "$execution_file" \
"$RUNNER_TEMP/review-diagnostics/diagnostics.json"

- name: Upload review failure diagnostics
if: ${{ always() && steps.claude.outcome == 'failure' }}
uses: actions/upload-artifact@v4
with:
name: review-failure-${{ github.run_id }}-${{ github.run_attempt }}
path: ${{ runner.temp }}/review-diagnostics/diagnostics.json
retention-days: 7
if-no-files-found: warn

- name: Record Claude end time
if: always()
run: echo "CLAUDE_END_NS=$(date +%s%N)" >> "$GITHUB_ENV"
Expand Down
67 changes: 67 additions & 0 deletions scripts/review-diagnostics.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
#!/usr/bin/env python3
"""Save terminal review errors without publishing tool transcripts."""
import html
import json
import os
from pathlib import Path
import sys

FIELDS = ('type', 'subtype', 'is_error', 'result', 'errors', 'duration_ms',
'num_turns', 'total_cost_usd', 'session_id')


def read_result(source):
try:
text = source.read_text()
try:
messages = json.loads(text)
except json.JSONDecodeError:
messages = [json.loads(line) for line in text.splitlines() if line.strip()]
except (OSError, ValueError):
return {'status': 'execution output unavailable'}
if isinstance(messages, dict):
messages = [messages]
if not isinstance(messages, list):
return {'status': 'result unavailable'}
results = [message for message in messages
if isinstance(message, dict) and message.get('type') == 'result']
if not results:
return {'status': 'result unavailable'}
return {'status': 'captured', 'result': {
key: results[-1][key] for key in FIELDS if key in results[-1]
}}


def redact(value, secrets):
if isinstance(value, str):
for secret in secrets:
value = value.replace(secret, '[REDACTED]')
return value
if isinstance(value, list):
return [redact(item, secrets) for item in value]
if isinstance(value, dict):
return {redact(key, secrets): redact(item, secrets) for key, item in value.items()}
return value


def main():
source, destination = map(Path, sys.argv[1:])
secrets = json.loads(os.environ.get('DIAGNOSTIC_SECRET_VALUES', '[]'))
secrets += [value for key, value in os.environ.items()
if key.startswith('DIAGNOSTIC_TOKEN_') and value]
secrets = sorted(set(filter(None, secrets)), key=len, reverse=True)
report = redact(read_result(source), secrets)
output = json.dumps(report, indent=2)
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_text(output + '\n')
summary = os.environ.get('GITHUB_STEP_SUMMARY')
if summary:
with open(summary, 'a') as stream:
stream.write('\n### AI review failure diagnostics\n\n')
stream.write('Terminal result only; tool transcripts are omitted. '
'The artifact contains the same sanitized details.\n\n')
stream.write('<pre>' + html.escape(output[:16000]) + '</pre>\n')


if __name__ == '__main__':
main()
65 changes: 65 additions & 0 deletions tests/test_review_diagnostics.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest

SCRIPT = Path(__file__).resolve().parents[1] / 'scripts/review-diagnostics.py'


class ReviewDiagnosticsTest(unittest.TestCase):
def run_diagnostics(self, content):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / 'execution.json'
if content is not None:
source.write_text(content)
destination = root / 'diagnostics.json'
summary = root / 'summary.md'
result = subprocess.run(
['python3', str(SCRIPT), str(source), str(destination)],
env={**os.environ, 'GITHUB_STEP_SUMMARY': str(summary),
'DIAGNOSTIC_SECRET_VALUES': json.dumps(['private-test-secret'])},
capture_output=True, text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
return json.loads(destination.read_text()), summary.read_text()

def test_retains_error_without_tool_transcript_or_secrets(self):
report, summary = self.run_diagnostics(json.dumps([
{'type': 'assistant', 'message': {'content': 'private tool transcript'}},
{'type': 'result', 'subtype': 'success', 'is_error': True,
'result': 'API Error: private-test-secret', 'num_turns': 36,
'errors': ['Provider failed'], 'extra': 'private tool transcript'},
]))
self.assertEqual(report['result']['result'], 'API Error: [REDACTED]')
self.assertEqual(report['result']['errors'], ['Provider failed'])
self.assertNotIn('private tool transcript', json.dumps(report) + summary)
self.assertNotIn('private-test-secret', json.dumps(report) + summary)
self.assertIn('Provider failed', summary)

def test_reads_json_lines(self):
report, _ = self.run_diagnostics('\n'.join([
json.dumps({'type': 'system'}),
json.dumps({'type': 'result', 'is_error': True, 'result': 'API error'}),
]))
self.assertEqual(report['result']['result'], 'API error')

def test_missing_and_malformed_output_remain_diagnostic(self):
for content in [None, 'invalid private-test-secret']:
with self.subTest(content=content):
report, summary = self.run_diagnostics(content)
self.assertIn('unavailable', report['status'])
self.assertNotIn('private-test-secret', summary)

def test_no_result_does_not_publish_transcript(self):
report, summary = self.run_diagnostics(json.dumps([
{'type': 'assistant', 'message': 'private tool transcript'},
]))
self.assertEqual(report['status'], 'result unavailable')
self.assertNotIn('private tool transcript', summary)


if __name__ == '__main__':
unittest.main()
Loading