grade.yml

name: Grade intent document
on:
  workflow_call:
    inputs:
      grader-sha:
        required: true
        type: string
permissions:
  contents: read
  pull-requests: read
  id-token: write
jobs:
  intent-grade:
    name: intent-grade
    runs-on: ubuntu-24.04
    timeout-minutes: 5
    steps:
      - name: Read and check immutable documents
        shell: python
        env:
          GITHUB_TOKEN: ${{ github.token }}
          SUBMISSION_REPO: ${{ github.repository }}
          SUBMISSION_SHA: ${{ github.event.pull_request.head.sha }}
          SUBMISSION_PR: ${{ github.event.pull_request.number }}
          SUBMISSION_EVENT: ${{ github.event_name }}
          GRADER_SHA: ${{ inputs.grader-sha }}
        run: |
          """Document-only checker. Candidate content is parsed as data, never executed."""
          import csv
          import io
          import json
          import re
          from pathlib import Path
          
          MAX_BYTES = 24576
          NAMES = ['Goal', 'Scope', 'Constraints', 'Plan', 'Acceptance', 'Verification']
          CATEGORIES = ['explicit-priority', 'default-priority', 'quoted-comma', 'empty-title',
                        'unknown-column', 'invalid-priority', 'row-limit']
          
          
          def predict(case):
              if not isinstance(case, dict) or set(case) - {'csv', 'expected', 'repeat'}:
                  raise ValueError('case shape')
              raw = case.get('csv')
              repeat = case.get('repeat', 1)
              if not isinstance(raw, str) or len(raw) > MAX_BYTES or type(repeat) is not int or not 1 <= repeat <= 101:
                  raise ValueError('case limits')
              try:
                  rows = list(csv.reader(io.StringIO(raw), strict=True))
              except csv.Error:
                  return {'error': 'unknown-column'}, {'unknown-column'}
              if not rows:
                  return {'error': 'unknown-column'}, {'unknown-column'}
              header, data = rows[0], rows[1:]
              if repeat != 1 and len(data) != 1:
                  raise ValueError('repeat requires one data row')
              data = data * repeat
              if len(data) > 100:
                  return {'error': 'row-limit'}, {'row-limit'}
              if ('title' not in header or len(set(header)) != len(header)
                      or set(header) - {'title', 'priority'} or any(len(row) != len(header) for row in data)):
                  return {'error': 'unknown-column'}, {'unknown-column'}
              tickets, tags = [], set()
              for row in data:
                  values = dict(zip(header, (v.strip() for v in row)))
                  if not values['title']:
                      return {'error': 'empty-title'}, {'empty-title'}
                  priority = values.get('priority', '')
                  if priority and priority not in ['low', 'normal', 'urgent']:
                      return {'error': 'invalid-priority'}, {'invalid-priority'}
                  tags.add('explicit-priority' if priority else 'default-priority')
                  if ',' in values['title']:
                      tags.add('quoted-comma')
                  tickets.append({'title': values['title'], 'priority': priority or 'normal'})
              return {'tickets': tickets}, tags
          
          
          def no_duplicate_keys(pairs):
              result = {}
              for key, value in pairs:
                  if key in result:
                      raise ValueError('duplicate JSON key')
                  result[key] = value
              return result
          
          
          def check(text, claude):
              prose = re.sub(r'^```[^\n]*\n.*?^```\s*$', '', text, flags=re.M | re.S)
              sections = re.findall(r'^## (Goal|Scope|Constraints|Plan|Acceptance|Verification)\s*\n(.*?)(?=^## |\Z)', prose, re.M | re.S)
              results = []
              def add(label, passed):
                  results.append({'check': label, 'passed': bool(passed)})
              add('Six substantive sections', len(sections) == 6 and {s[0] for s in sections} == set(NAMES)
                  and all(len(re.sub(r'\s+', ' ', s[1]).strip()) >= 30 for s in sections))
              add('No placeholders', not re.search(r'\b(TODO|TBD|FIXME)\b|<insert|\[your .+?\]', text, re.I))
              loader = re.sub(r'^```[^\n]*\n.*?^```\s*$', '', claude, flags=re.M | re.S)
              loader = re.sub(r'`[^`]*`', '', loader)
              add('Claude imports intent', bool(re.search(r'(?<!\S)@(?:\./)?intent\.md\b', loader)))
              blocks = re.findall(r'^```json\s*\n(.*?)^```\s*$', text, re.M | re.S)
              contract = {}
              try:
                  if len(blocks) == 1:
                      contract = json.loads(blocks[0], object_pairs_hook=no_duplicate_keys)
                      if not isinstance(contract, dict):
                          contract = {}
              except (ValueError, RecursionError):
                  pass
              keys = {'schema','change','allowed_changes','protected_paths','non_goals','requirements','verification','approval','acceptance_cases'}
              add('Contract identity and shape', set(contract) == keys and type(contract.get('schema')) is int
                  and contract['schema'] == 1 and contract.get('change') == 'ticket-csv-preview')
              add('Allowed changes', contract.get('allowed_changes') == ['src/import-tickets.mjs','tests/import-tickets.test.mjs'])
              add('Protected paths', contract.get('protected_paths') == ['src/tickets.mjs','.github/','CLAUDE.md'])
              add('Non-goals', contract.get('non_goals') == ['ui','database','network'])
              req = contract.get('requirements', {})
              if not isinstance(req, dict):
                  req = {}
              add('CSV columns', req.get('columns') == ['title','priority'])
              add('Priority rules', req.get('default_priority') == 'normal' and req.get('valid_priorities') == ['low','normal','urgent'])
              add('Validation and preview constraints', req.get('reject_unknown_columns') is True
                  and req.get('reject_empty_title') is True and type(req.get('max_rows')) is int and req['max_rows'] == 100
                  and req.get('side_effects') == 'preview-only'
                  and set(req) == {'columns','default_priority','valid_priorities','reject_unknown_columns','reject_empty_title','max_rows','side_effects'})
              add('Verification commands', contract.get('verification') == ['npm test','git diff --check'])
              add('Approval boundary', contract.get('approval') == 'ask-before-implementation')
              cases = contract.get('acceptance_cases', [])
              valid = isinstance(cases, list) and 1 <= len(cases) <= 20
              covered = set()
              if valid:
                  for case in cases:
                      try:
                          expected, tags = predict(case)
                          # JSON comparison distinguishes booleans from integers.
                          correct = json.dumps(case.get('expected'), sort_keys=True) == json.dumps(expected, sort_keys=True)
                          valid = valid and correct
                          if correct:
                              covered.update(tags)
                      except (ValueError, TypeError, RecursionError):
                          valid = False
              add('All acceptance examples are correct', valid)
              for category in CATEGORIES:
                  add('Example: ' + category, category in covered)
              return results
          
          
          def report(text, claude, submission_sha, grader_sha, invalid=False):
              results = check(text, claude) if not invalid else []
              passed = sum(item['passed'] for item in results)
              score = passed * 5
              return {'schema': 1, 'exercise': 'intent-md', 'grader_sha': grader_sha,
                      'submission_sha': submission_sha, 'status': 'invalid' if invalid else 'graded',
                      'score': score, 'max': 100,
                      'points': {'scope': 0, 'static': 0, 'tests': score, 'regression': 0},
                      'gates': {'tests': {'total': 20, 'passed': passed}}}, results
          
          
          def read_document(path):
              path = Path(path)
              if path.is_symlink() or not path.is_file() or path.stat().st_size > MAX_BYTES:
                  raise ValueError('document must be a regular file no larger than 24 KiB')
              return path.read_bytes().decode('utf-8', errors='strict')
          
          
          
          MANIFEST = {'.gitignore': '1d825c27711bf4dbc3566886d56bfe1f0b3b30ec', 'CLAUDE.md': '40fef0245898500cb693485649e73e0eea811844', 'README.md': '3a1dd680186a69c48ff92ea63eb89fbf4577fb72', 'docs/product-brief.md': '155581e8bffc457ec2249257230442df942e705e', 'fixtures/sample.csv': '961ca0b5cdf04a40ceb47bbca983dc553733c75e', 'package.json': '842be2aa26c92efc3f4aaf4828d133d6491593c5', 'src/tickets.mjs': 'dde96a94663a65f0e14547558747a1c3e5109baa', 'tests/tickets.test.mjs': 'cae4a4c26a811392c784d6e0389e16eff9ebcd8e'}
          """GitHub runner entrypoint; fetch only immutable blobs, never checkout candidate code."""
          import base64
          import hashlib
          import os
          import urllib.request
          
          
          def api(path):
              req = urllib.request.Request('https://api.github.com/' + path, headers={
                  'Authorization': 'Bearer ' + os.environ['GITHUB_TOKEN'],
                  'Accept': 'application/vnd.github+json', 'X-GitHub-Api-Version': '2022-11-28'})
              with urllib.request.urlopen(req, timeout=20) as response:
                  raw = response.read(1048577)
                  if len(raw) > 1048576:
                      raise RuntimeError('GitHub response limit')
                  return json.loads(raw)
          
          
          def caller_workflow(sha):
              return ('name: Check intent submission\non:\n  pull_request:\n    types: [opened, synchronize, reopened]\n    branches: [main]\n'
                      'permissions:\n  contents: read\n  pull-requests: read\njobs:\n  grade:\n'
                      '    uses: TensorOpsAI/tensorlearn-intent-grader/.github/workflows/grade.yml@' + sha + '\n'
                      '    with:\n      grader-sha: ' + sha + '\n')
          
          
          def remote_documents(repo, sha, grader_sha, manifest, pr):
              tree = api(f'repos/{repo}/git/trees/{sha}?recursive=1')
              if tree.get('truncated') or len(tree.get('tree', [])) > 100:
                  raise ValueError('submission tree limit')
              blobs = {item['path']: item for item in tree['tree'] if item['type'] != 'tree'}
              allowed = set(manifest) | {'intent.md', '.github/workflows/grade.yml'}
              if set(blobs) != allowed:
                  raise ValueError('only intent.md may change')
              for path, item in blobs.items():
                  if item.get('mode') != '100644' or item.get('type') != 'blob':
                      raise ValueError('regular files only')
                  if path in manifest and item['sha'] != manifest[path]:
                      raise ValueError('starter files changed')
              def read(path):
                  item = blobs[path]
                  if item.get('size', MAX_BYTES+1) > MAX_BYTES:
                      raise ValueError('document limit')
                  blob = api(f'repos/{repo}/git/blobs/{item["sha"]}')
                  if blob.get('encoding') != 'base64':
                      raise ValueError('unsupported blob')
                  data = base64.b64decode(blob['content'])
                  if len(data) > MAX_BYTES:
                      raise ValueError('document limit')
                  expected = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest()
                  if expected != item['sha']:
                      raise ValueError('blob identity mismatch')
                  return data.decode('utf-8', errors='strict')
              if read('.github/workflows/grade.yml') != caller_workflow(grader_sha):
                  raise ValueError('caller workflow changed')
              # Read the exact open PR in the caller repo. Never grade an upstream/cross-repo PR.
              pull = api(f'repos/{repo}/pulls/{pr}')
              if not (pull.get('number') == pr and pull.get('state') == 'open'
                      and not pull.get('merged_at')
                      and pull.get('head', {}).get('sha') == sha
                      and pull.get('base', {}).get('ref') == 'main'
                      and (pull.get('base', {}).get('repo') or {}).get('full_name', '').lower() == repo.lower()
                      and (pull.get('head', {}).get('repo') or {}).get('full_name', '').lower() == repo.lower()):
                  raise ValueError('no matching open PR inside this repository')
              return read('intent.md'), read('CLAUDE.md')
          
          
          def run():
              sha, grader_sha, repo = os.environ['SUBMISSION_SHA'], os.environ['GRADER_SHA'], os.environ['SUBMISSION_REPO']
              if not re.fullmatch('[a-f0-9]{40}', sha) or not re.fullmatch('[a-f0-9]{40}', grader_sha):
                  raise RuntimeError('invalid immutable revision')
              if not re.fullmatch(r'[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+', repo):
                  raise RuntimeError('invalid repository')
              if os.environ.get('SUBMISSION_EVENT') != 'pull_request':
                  raise RuntimeError('only pull-request checks are supported')
              pr = int(os.environ['SUBMISSION_PR'])
              if pr < 1:
                  raise RuntimeError('invalid PR number')
              try:
                  text, claude = remote_documents(repo, sha, grader_sha, MANIFEST, pr)
                  score, checks = report(text, claude, sha, grader_sha)
              except (ValueError, UnicodeError):
                  score, checks = report('', '', sha, grader_sha, invalid=True)
              # Transport/API failures propagate and produce no fabricated score.
              Path('result').mkdir(exist_ok=True)
              Path('result/score.json').write_text(json.dumps(score) + '\n')
              with open(os.environ['GITHUB_STEP_SUMMARY'], 'a') as summary:
                  summary.write(f'# Intent document: {score["score"]}/100 ({score["status"]})\n\n')
                  if not checks:
                      summary.write('Submission rejected: verify a matching open PR in your own fork, unchanged starter files, regular UTF-8 documents, and the size limit.\n')
                  for item in checks:
                      summary.write(('- PASS: ' if item['passed'] else '- NEEDS WORK: ') + item['check'] + '\n')
                  summary.write('\nThis is a document contract score. Server ingestion is a separate check.\n')
          
          
          if __name__ == '__main__':
              run()
      - name: Upload score
        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02
        with:
          name: intent-score
          path: result/score.json
          if-no-files-found: error
          retention-days: 7
      - name: Report score to TensorLearn
        continue-on-error: true
        env:
          REPORT_URL: https://tensorops.ai/api/github/report
          AUDIENCE: https://tensorops.ai/code-labs
        run: |
          token=$(curl -sSf -H "Authorization: Bearer $ACTIONS_ID_TOKEN_REQUEST_TOKEN" \
            "$ACTIONS_ID_TOKEN_REQUEST_URL&audience=$AUDIENCE" | jq -r .value)
          echo "::add-mask::$token"
          for wait in 2 4 8 16 32; do
            status=$(curl -sS -o "$RUNNER_TEMP/report.txt" -w '%{http_code}' -X POST "$REPORT_URL" \
              -H "Authorization: Bearer $token" -H 'Content-Type: application/json' \
              --data-binary @result/score.json || echo 000)
            echo "report: HTTP $status $(cat "$RUNNER_TEMP/report.txt")"
            case "$status" in
              2*) exit 0 ;;
              429|5*|000) sleep "$wait" ;;
              *) exit 1 ;;
            esac
          done
          exit 1