# -*- coding: utf-8 -*- """The escalation path: when code cannot compute it, a small model looks. The ladder this implements, ruled 2026-08-30: code can compute it -> code computes it code cannot -> Haiku reads the drawing Haiku cannot settle it -> NOT RUN or WARN -> a human cleans up and the rule that governs everything below it, `docs/ticket-checklist.md` section 8bis: **a check whose verdict comes from a model looking at a rendered sheet reports NOT RUN, and raises a WARN alongside it.** Not a pass, not a fail, and never silent. The report may say what it could not judge and that somebody should look. It may not paraphrase the content, score it, or imply it was reviewed. WHY THESE FINDINGS ARE KEPT IN THEIR OWN LIST --------------------------------------------- A model-read verdict and a computed verdict fail differently. A wrong computed answer is a bug -- reproducible, findable, fixable. A wrong read is a hallucination -- it may not reproduce, and there is nothing to fix. A reader who cannot tell them apart will trust the second as much as the first, and that is the whole risk of this path. So they never share a container. `run_all`'s findings are computed; these come back from `read_for_ticket` as a SEPARATE list, they are the only findings in the system whose codes begin `AI-`, they own no row in `CHECKLIST_ITEMS`, and they carry `source='model-read'` on the object itself. Four independent markers, because one of them will eventually be dropped by an edit somewhere and the other three still have to hold. WHY EVERY OUTCOME IS A WARN --------------------------- All three answers the skill allows -- `clear`, `problem_seen`, `cannot_tell` -- come back as WARN. `clear` is not a pass: the model looked at one view and saw nothing wrong, which is a reason for a person to look and not a reason to skip looking. Section 8bis allows exactly one shape (NOT RUN on the row, WARN beside it) and this is it. IronPython 2.7: no f-strings, no pathlib. """ # pylint: disable=import-error,invalid-name,broad-except import io import json import os import time from vibe import knowledge as K from vibe import ticketpage as TP # `vibe.ticketpage` is Revit-free at module scope on purpose -- it owns the # checklist table and the state words, and this file has to walk both to # write the run's disposition without ever being able to invent a row. # # `vibe.ticketqc` imports the Revit API at module scope, so it is imported # lazily, inside `to_findings` alone. Everything else in this file is plain # dicts and arithmetic and runs on host CPython under the test suite -- which # is the point: the verdict is not test-pinnable, so the plumbing around it # has to be. # ------------------------------------------------------------------ the skill SKILL_FILENAME = 'ticketqc-vision.skill.md' def skill_path(): """The instructions the model reads, versioned with the code. A file, not a string in a function: a prompt buried in a function is a prompt nobody reviews, and this one is the only thing standing between a small model and a confident wrong answer about a piece of concrete. """ here = os.path.dirname(os.path.abspath(__file__)) root = os.path.dirname(os.path.dirname(os.path.dirname(here))) return os.path.join(root, 'docs', SKILL_FILENAME) # ----------------------------------------------------------- how big to render # # The evidence that resolution decides correctness is in QNA/, and it is not # theoretical: # # * a PDF text layer returned `1316"` for BOTH `13/16"` and `1 3/16"`, an # inch apart, and only a render separated them (the section-11000 extraction in QNA/); # * a phantom strand position survived four sheets until somebody rendered # at 450 dpi; # * approval stamps invisible to text search were found only by looking, on # 22 consecutive sheets whose text layer returned thirteen characters. # # So the resolution is the check. What decides it here is a hard ceiling on # the far end: the API downscales any image whose long edge exceeds 1568 px, # so a render bigger than that is thrown away before the model sees it. The # model gets about 1.1 megapixels and no more, whatever we send. # # That single fact settles the design. A 34 x 22 in sheet rendered whole to # 1500 px is 44 dpi -- illegible, and no bigger render can help, because the # extra pixels are discarded on arrival. **So we never render a sheet. We # render one VIEW**, and the effective resolution is then set by how much # paper that view occupies: # # effective dpi = long edge in pixels / the view's long side in inches # # A 6-inch note view at 1500 px is 250 dpi. A 20-inch bed plan is 75 dpi and # is refused rather than guessed at, which is the same discipline the skill # asks of the model. LONG_EDGE_PX = 1500 # just under the 1568 px ceiling, so nothing is # resampled on the way in MIN_EFFECTIVE_DPI = 150.0 # below this the QNA evidence says a fraction or # a bar mark is not reliably separable, and a # render nobody can read is a NOT RUN, not a # question def render_plan(paper_w_in, paper_h_in, long_edge_px=LONG_EDGE_PX, floor_dpi=MIN_EFFECTIVE_DPI): """How big to render one view, and whether it is worth rendering at all. paper_w_in / paper_h_in -- the view's printed size on the sheet, inches. Returns a dict: pixels (long edge), effective_dpi, ok, reason. `ok` False means do not call the model -- the row is NOT RUN with `reason`, which is the honest outcome and costs nothing. Pure arithmetic, host-testable, no Revit. """ try: w = float(paper_w_in) h = float(paper_h_in) except (TypeError, ValueError): return {'pixels': None, 'effective_dpi': None, 'ok': False, 'reason': 'the view\'s printed size on the sheet could not ' 'be measured, so no legible render could be sized'} longest = max(w, h) if longest <= 0: return {'pixels': None, 'effective_dpi': None, 'ok': False, 'reason': 'the view has no printed extent on the sheet'} dpi = float(long_edge_px) / longest if dpi < floor_dpi: return { 'pixels': int(long_edge_px), 'effective_dpi': dpi, 'ok': False, 'reason': ('this view prints {0:.1f} in across, so the largest ' 'render the model can actually receive ({1} px on the ' 'long edge -- bigger is downscaled on arrival) is only ' '{2:.0f} dpi, under the {3:.0f} dpi floor at which a ' 'stacked fraction or a bar mark stops being separable' .format(longest, int(long_edge_px), dpi, floor_dpi))} return {'pixels': int(long_edge_px), 'effective_dpi': dpi, 'ok': True, 'reason': None} # ------------------------------------------------------- Revit: view -> a PNG def viewport_paper_size_in(viewport): """A placed view's printed size on its sheet, in inches. Sheet coordinates are feet, so this is the drawn size on paper -- exactly what decides the effective resolution. Returns (w_in, h_in), or (None, None) when the viewport will not give up its outline. """ try: box = viewport.GetBoxOutline() lo, hi = box.MinimumPoint, box.MaximumPoint return (abs(hi.X - lo.X) * 12.0, abs(hi.Y - lo.Y) * 12.0) except Exception: return (None, None) def collect_viewports(doc, sheets): """Every placed viewport on this ticket's sheets, as plain dicts. Its own collector rather than a new key on `ticketqc.read_ticket`: this is needed only when an escalation actually runs, and every ticket in a 1105-sheet model would otherwise pay for it. Read-only -- a collector opens no transaction. """ from Autodesk.Revit.DB import FilteredElementCollector, Viewport out, seen = [], set() for sheet in (sheets or []): try: found = FilteredElementCollector(doc, sheet.Id) \ .OfClass(Viewport).ToElements() except Exception: continue for vp in found: try: # .IntegerValue is deprecated for .Value in Revit 2024+, and # both bindings are in the field. Read it defensively. try: key = vp.Id.IntegerValue except Exception: key = vp.Id.Value if key in seen: continue seen.add(key) view = doc.GetElement(vp.ViewId) if view is None: continue out.append({'name': view.Name, 'view_id': vp.ViewId, 'viewport': vp}) except Exception: continue return out def render_view_png(doc, view_id, out_dir, long_edge_px=LONG_EDGE_PX): """Export one view to a PNG. Returns its path, or raises. `Document.ExportImage` is a READ operation -- no Transaction is opened, nothing in the model is written, and nothing in this module ever will be. The production model is read-only and this path does not change that. Revit decorates the filename with the view's own name, so the file is found by looking at what appeared rather than by assuming the name. """ from Autodesk.Revit.DB import (ImageExportOptions, ExportRange, ImageFileType, ZoomFitType, FitDirectionType, ElementId) from System.Collections.Generic import List if not os.path.isdir(out_dir): os.makedirs(out_dir) stem = os.path.join(out_dir, 'view_{0}'.format(int(time.time() * 1000))) before = set(os.listdir(out_dir)) opts = ImageExportOptions() opts.ExportRange = ExportRange.SetOfViews ids = List[ElementId]() ids.Add(view_id) opts.SetViewsAndSheets(ids) opts.FilePath = stem opts.HLRandWFViewsFileType = ImageFileType.PNG opts.ShadowViewsFileType = ImageFileType.PNG opts.ZoomType = ZoomFitType.FitToPage opts.FitDirection = FitDirectionType.Horizontal opts.PixelSize = int(long_edge_px) doc.ExportImage(opts) made = [f for f in os.listdir(out_dir) if f not in before and f.lower().endswith('.png')] if not made: raise IOError('Revit exported no image for this view') return os.path.join(out_dir, sorted(made)[0]) # ------------------------------------------------------- what escalates, and why # # NOT everything. Escalation is for checks where code has already given up and # the checklist says so in writing. Anything a comparison can settle stays a # comparison -- a model asked a question code could answer is a hallucination # risk taken for nothing. # # The proposed list, in the order they are worth doing: # # 5.5 bar marks legible and unique on the drawing -- BUILT (below). # Grade D, "a person reads the drawing", and the only checklist row # that has always rendered FOR ENGINEER because there was nothing # else to render. Legibility is not a property of the model; it is a # property of the picture, so a picture is the only thing that can # answer it. # # 6.5r a note view that says SOMETHING -- but is it the RIGHT something? # DEFINED in the skill (`NOTE-SENSE`), NOT WIRED. Emptiness is # already counted exactly by `ticketpage.note_view_content_status`; # only the residue is left, and section 8bis's paraphrase ban bites # hardest here -- the model may say a note reads like a placeholder # and may not say what it says. Left for a second pass because it # needs `notes.instruction_views` populated to know which views to # render, and no plant has populated it yet, so it would report NOT # RUN on every ticket today anyway. # # 5.3 face designations (N.F./F.F./E.L./S.I.F.) against the model side. # NOT PROPOSED as a vision check. Grade C: code narrows it to a # shortlist and a person decides. The face a callout names is legible # text, but whether it is RIGHT needs the model's geometry, which the # picture does not carry -- rule 6 of the skill would make every # answer `cannot_tell`. # # 8.1 unresolved design-summary lines. NOT PROPOSED. Already grade C with # a computed shortlist; a picture adds nothing. # # The title block, the BOM, the handling dimensions. NOT PROPOSED, and # this one is worth stating because it is the tempting one: every number # on them is already read by code from the model, exactly, and asking a # model to re-read them off a picture would replace an exact answer with a # guess. Escalation goes DOWN the ladder, never up it. ESCALATIONS = ( { 'item': '5.5', 'question_id': 'BARMARK-LEGIBLE', 'code': 'AI-BARMARK', 'what': 'whether the reinforcing bar marks on the drawing are ' 'legible and each distinct from the others', # Which views carry the bar marks is plant nomenclature, exactly like # `notes.instruction_views`. Default empty -> the check reports NOT # RUN naming that, and never a pass on a ticket nobody looked at. 'views_key': 'vision.barmark_views', }, ) def escalation(item): for row in ESCALATIONS: if row['item'] == item: return row return None def _plant_views(row, know): """The plant's view names for one escalation. Spelled out one literal at a time rather than `K.get(know, row['views_key'])`, and that is not clumsiness: `tools/knowledge_drift.py` scans for literal key reads, and a key reached through a variable reads to it as DECLARED BUT NEVER READ -- a key someone can set that the tool appears not to honour. A knowledge key nobody can see is a key that quietly stops working. """ if row['views_key'] == 'vision.barmark_views': return K.get(know, 'vision.barmark_views', []) or [] return [] # ------------------------------------------------------------ the run options # # Cost and latency are real and are handled where the precedent already is: # run-time options (`configmodel.RUN_OPTIONS`), not the knowledge tree. # Whether THIS RUN is willing to spend money on model calls is a property of # the run, not a fact about the plant. # # Unlike `expensive_geometry`, this one is OFF by default, and the difference # is deliberate. Reading a concrete solid costs seconds of a machine nobody is # billing; a model call costs money and leaves the building. Off-by-default # prints NOT RUN with a reason on every escalated row, which is honest and # free, and an escalation nobody asked for is exactly the "silently expensive" # outcome to avoid. OPTION_ON = 'vision_escalation' OPTION_MAX = 'vision_escalation_max_calls' VISION_OFF_REASON = ( 'the run-time option vision_escalation is OFF for this run, so no ' 'drawing was read by a model -- turn it on in the project config ' '(run-options.yaml) to escalate this check') def vision_off_reason(): """One sentence, said one way, wherever a check has to explain itself.""" return VISION_OFF_REASON def run_settings(root_dir, options=None): """(on, max_calls) for this run. A broken options file is OFF, not ON.""" if options is None: try: from vibe import configmodel options = configmodel.load_run_options(root_dir) except Exception: options = {} on = bool((options or {}).get(OPTION_ON, False)) try: cap = int((options or {}).get(OPTION_MAX, 25)) except (TypeError, ValueError): cap = 25 return on, max(0, cap) # --------------------------------------------------- building the request file def build_request(items, max_calls=25, model=None, skill=None): """The JSON handed to `tools/vision_read.py`. The two halves talk through files, never imports: IronPython 2.7 in here, CPython 3.12 over there, and the boundary is why this codebase works. """ req = {'skill': skill or skill_path(), 'max_calls': int(max_calls), 'items': list(items)} if model: req['model'] = model return req def write_request(path, request): text = json.dumps(request, indent=2, sort_keys=True) if not isinstance(text, type(u'')): text = text.decode('utf-8') with io.open(path, 'w', encoding='utf-8') as fh: fh.write(text) return path def read_response(path): """The response file, or None. A file that is not there is not a pass.""" try: with io.open(path, 'r', encoding='utf-8') as fh: return json.loads(fh.read()) except Exception: return None # ------------------------------------------------- response -> WARN findings def _clip(text, limit=160): text = (text or u'').strip().replace(u'\n', u' ') if len(text) > limit: return text[:limit] + u'...' return text LEAD = u'[read by a model, not computed] ' # The tail says "did not run" rather than naming a status word, because the # rows this path touches carry two of them: 5.5 is grade D and reads FOR # ENGINEER, and anything wired later would read NOT RUN. Both mean the same # thing here -- nothing was computed -- and a tail that named only one of # them would be wrong on the row it was actually built for. TAIL = (u' -- NOT a verdict; the checklist row did not run and a person ' u'still has to look') def verdict_row(result, what=None, view=None): """One model-read result -> one WARN row carrying its provenance. Every outcome is a WARN. `clear` included: the model looked at one view and saw nothing wrong, which is a reason for a person to look, never a reason to skip looking. Section 8bis allows exactly one shape and this is it -- and there is no branch here that can produce 'OK' or 'FAIL'. The message always OPENS with what read it, because a reader who skims one line has to get the provenance in that line. A plain dict, not a Finding: `vibe.ticketqc` needs Revit at import time and the honesty rules encoded here have to be testable without it. """ code = result.get('code') or 'AI-UNKNOWN' where = view or result.get('view') subject = what or 'this' if result.get('status') != 'OK': body = (u'a model was asked to look at {0}{1} and could not: {2}' .format(subject, u' on ' + where if where else u'', result.get('reason') or u'no reason given')) else: answer = result.get('answer') why = _clip(result.get('why')) if answer == 'problem_seen': body = (u'a model looked at {0}{1} and reports something wrong{2}' u'{3}'.format(subject, u' on ' + where if where else u'', u' (' + _clip(result.get('where'), 60) + u')' if result.get('where') else u'', u': ' + why if why else u'')) elif answer == 'clear': body = (u'a model looked at {0}{1} and saw nothing wrong{2}' .format(subject, u' on ' + where if where else u'', u' -- ' + why if why else u'')) else: body = (u'a model looked at {0}{1} and could not tell{2}' .format(subject, u' on ' + where if where else u'', u' -- ' + why if why else u'')) return { 'severity': 'WARN', # the only severity this path emits 'code': code, 'source': 'model-read', 'message': LEAD + body + TAIL, 'model_read': { 'question_id': result.get('question_id'), 'answer': result.get('answer'), 'status': result.get('status') or 'NOT_RUN', 'reason': result.get('reason'), 'why': result.get('why'), 'where': result.get('where'), 'view': where, 'model': result.get('model'), 'effective_dpi': result.get('effective_dpi'), }, } def rows_from_response(response, subjects=None): """Every result -> a WARN row. A missing response is NOT RUN, never a pass. `subjects` maps a result id to the phrase describing what was asked and which view was rendered, so the message reads as a sentence. """ subjects = subjects or {} if not response: return [] out = [] for result in response.get('results') or []: sub = subjects.get(result.get('id')) or {} out.append(verdict_row(result, what=sub.get('what'), view=sub.get('view'))) return out def not_run_row(code, what, reason, view=None): """The escalation did not happen at all. Still a WARN beside a NOT RUN. Used for every reason short of the model itself: the switch is off, the plant has not named the views, the render would have been illegible, the budget was spent, the response file never appeared. Never silence, because a silent row is read as a pass by everybody who skims a report. """ return verdict_row({'status': 'NOT_RUN', 'reason': reason, 'code': code}, what=what, view=view) def to_findings(rows): """Rows -> `ticketqc.Finding` objects, provenance stamped on each. The only place this module touches `vibe.ticketqc`, and it is lazy because that module needs the Revit API at import time. The `source` field is the object-level marker; the `AI-` code prefix, the separate list and the separate report section are the other three. """ from vibe.ticketqc import Finding out = [] for row in rows or []: f = Finding(row['severity'], row['code'], row['message'], source=row.get('source', 'model-read')) f.model_read = row.get('model_read') out.append(f) return out # ------------------------------------------------------------- the JSON shape DISCLAIMER = ( 'Every finding under model_read came from a small model looking at a ' 'rendered view. It was not computed, it is not reproducible, it is not a ' 'verdict, and no checklist row passes or fails on it. The matching ' 'checklist rows still say nothing was computed -- FOR ENGINEER on a ' 'grade D item, NOT RUN elsewhere. A person still has to look.') # 'qc-run/1' -> 'qc-run/2' earlier the same day: the NOT APPLICABLE state, a # `panel_type` block per piece, and a `not_applicable` block on any row a # knowledge layer switched off. # BUMPED 'qc-run/2' -> 'qc-run/3' on 2026-08-31. A plant may now renumber its # own checklist (`checklists.profiles`), so every row carries BOTH numbers: # `item` is the MASTER ID -- the stable identity every mute, rule and ruling # names -- and `display_number` is what that plant's board called it. A # consumer that had been reading `item` as "the number on the report" would # have been ambushed by a renumbering it could not see, which is exactly the # reason for the bump: discard whole on mismatch, never migrate. RUN_SCHEMA = 'qc-run/3' # ============================================================================ # WHY THE DISPOSITION IS IN THIS FILE AND NOT ONLY THE PAGE # ============================================================================ # # `qc-run.json` used to hold `findings` + `model_read` AND NOTHING ELSE. # Every checklist row, its state, and above all the REASON a row did not run # existed only in the HTML page and in `last-ticketqc.txt` -- both of them # written for a person to read. So the machine-readable output reproduced, # one layer down, the exact failure this tool exists to prevent: # # a consumer reading that file could not tell a row that PASSED from a # row NOBODY ASKED. Both looked like an absence of findings. # # NOT RUN is never a pass. A run whose 6.2 could not be judged and a run # whose 6.2 came back clean produced the same `qc-run.json`, and the second # one is a claim the first has no right to make. `checklist_states` has # refused that inference on the page since 08-28; the file was still making # it. This is the file half of the same rule. # # THREE THINGS FOLLOW FROM THAT, and each is a property of the code below # rather than a convention somebody has to remember: # # 1. THE ROWS COME FROM THE CHECKLIST TABLE, NOT FROM THE RUN. # `checklist_rows_json` walks `ticketpage.CHECKLIST_ITEMS` and emits an # entry for every number in it, then fills each one in from whatever the # run produced. A row the run never mentioned therefore cannot be # absent -- it lands with `reported_by_run: false` and a reason saying # the run produced no row for it, which is a defect in the tool and # reads as one. There is NO hand-kept list of row numbers anywhere in # this file. A hand-kept list is how 5.3 spent a day printing the wrong # state (`ticketpage.CHECKLIST_FOR_ENGINEER`), and the fix there was the # same fix as here: derive it from the table. # # 2. THE STATE IS A KEY, NOT A COLOUR. `ticketpage.state_key` gives every # state its own machine name, so NOT RUN and FAIL are distinguishable in # the data even though the page paints both red, and `judged` says # outright whether anything produced a verdict. # # 3. THE REASON IS VERBATIM AND UNCUT. `checklist_states` stopped capping # notes on 2026-08-30 (the owner: "不要有截断") and nothing here puts a # cap back. Whatever the row said, the file carries the whole of it. # # THE EVIDENCE FILE IS NOT THIS FILE. `ticketqc`'s `ticketqc/1` evidence # document carries a run's INPUTS and deliberately no verdicts, because # `qc-run.json` owns verdicts. That split stays clean: nothing measured moves # here, and nothing judged moves there. What the two now share is the # ENVELOPE -- `ticketqc.run_envelope` builds it for both, so provenance has # one shape in this codebase and not two. def qc_json(findings, model_read=None, tickets=None, envelope=None): """The run, machine-readable: the disposition AND the two kinds of finding. `findings` are computed (Finding objects from `run_all`). `model_read` came off a picture (rows from this module). They are not merged, not sorted together and not counted together -- a consumer that reads only `findings` gets exactly the computed result and cannot accidentally treat a read as one. Every entry on both sides carries `source` and a boolean `computed`, so a consumer that DOES merge them still cannot lose the distinction. That quarantine is older than this function's other half and is untouched by it. `tickets` is the new half: one entry per piece, each carrying EVERY checklist row with its state and its reason, plus `run_all`'s own disposition log and this job's mutes. Without it the file says which findings were raised and says nothing about which questions were asked, which is the same as saying they all were. `envelope` is `ticketqc.run_envelope(RUN_SCHEMA, doc)`. It is passed in rather than built here because `vibe.ticketqc` needs the Revit API at import time and this module must stay importable without it. A caller that passes none still gets a schema and a `missing` entry saying the provenance was never gathered -- absence is not a value here either. """ document = dict(envelope or _no_envelope()) document['schema'] = RUN_SCHEMA rows = [_ticket_json(t) for t in (tickets or [])] document['ticket_count'] = len(rows) document['tickets'] = rows document['checklist_states'] = [ {'state': state, 'state_key': key, 'judged': judged} for (state, key, judged) in TP.STATE_VOCABULARY] document['findings'] = [_json_row(f, 'computed') for f in (findings or [])] document['model_read'] = [_json_row(f, 'model-read') for f in (model_read or [])] document['model_read_disclaimer'] = DISCLAIMER if not rows: document.setdefault('missing', {})['tickets'] = ( 'this run recorded no per-piece disposition, so the findings ' 'below cannot be read as a checklist -- a row absent from a ' 'finding list is not a row that passed') return document def _no_envelope(): """An envelope for a caller that gathered no provenance. Every field null AND a sentence saying so. Silently omitting the keys would let a consumer read the file as though it had been written by a build nobody can name against knowledge nobody can list, and not notice. """ return { 'schema': RUN_SCHEMA, 'written_at': None, 'written_by': None, 'build': None, 'doc_key': None, 'model_stem': None, 'knowledge_sources': None, 'missing': {'envelope': ( 'no provenance was gathered for this run -- the build, the ' 'document and the knowledge layers in force are unknown, so ' 'nothing here can be re-judged against what produced it')}, } # ------------------------------------------------------ the per-piece entry def _ticket_json(entry): """One piece's disposition. `entry` is the button's run result. Reads `mark` / `design` / `spec_source` / `rows` / `rows_unmuted` / `run_log` / `muted_notes` -- the same keys the report is rendered from, so the file and the page cannot describe different runs. """ entry = entry or {} ticket = entry.get('ticket') or {} missing = {} mark = ticket.get('mark') if isinstance(ticket, dict) else None if not mark: missing['mark'] = 'this piece has no readable mark' design = entry.get('design') if not design: missing['design'] = 'this piece was not judged under a design number' spec_source = entry.get('spec_source') if not spec_source: missing['spec_source'] = ( 'nothing recorded whether the design intent came from a reviewed ' 'spec or from the weaker regex reading of the summary view') log = entry.get('run_log') if not log: missing['run_log'] = ( "the checks' own disposition log is empty, so every row below " 'falls back to the checklist item\'s own reason') # WHAT KIND OF PIECE THIS WAS TAKEN TO BE, with the basis. A row that # was not asked because "this does not apply to an insulated panel" is # only auditable beside the answer to "and was this one insulated, and # how was that decided". A consumer holding the rows without the # classification has the conclusion and not the premise. classification = (entry.get('ticket') or {}).get('panel_type') \ if isinstance(entry.get('ticket'), dict) else None if not isinstance(classification, dict): classification = None missing['panel_type'] = ( 'this run recorded no panel-type classification for the piece, ' 'so any row it declared not applicable cannot be audited against ' 'what the piece was taken to be') return { 'mark': mark, 'design': design, 'spec_source': spec_source, 'member_kind': _member_kind_json( (entry.get('ticket') or {}).get('member_kind') if isinstance(entry.get('ticket'), dict) else None), 'checklist_profile': _profile_json(entry.get('profile'), entry.get('off_board')), 'panel_type': _panel_type_json(classification), 'rows': checklist_rows_json(entry.get('rows'), unmuted_rows=entry.get('rows_unmuted'), log=log, muted_notes=entry.get('muted_notes'), applicability=entry.get( 'applicability_notes')), 'run_log': _log_json(log), 'muted': _muted_json(entry.get('muted_notes')), 'not_applicable': _applicability_json( entry.get('applicability_notes')), 'applicability_refused': _refusal_json( entry.get('applicability_problems')), 'missing': missing, } def _member_kind_json(member): """What kind of member the mark said this is. Never absent. `member_kind: null` is a real answer -- the prefix was unreadable or is not one the plant has declared -- and it travels with the sentence that says which, because it is the premise of the whole profile selection below it. """ member = member if isinstance(member, dict) else None if not member: return {'member_kind': None, 'prefix': None, 'why': None, 'read': False} return {'member_kind': member.get('member_kind'), 'prefix': member.get('prefix'), 'why': member.get('why'), 'read': True} def _profile_json(profile, off_board_ids=None): """WHICH BOARD RAN. The premise of every display number in `rows`. A file that carried renumbered rows without saying whose numbering it was would be unreadable a year from now -- and unauditable today, since the numbers would not match any board anybody could look up. `off_board` is the list the report also shows: master ids this profile never named, which were ASKED ANYWAY. It is a fact about the YAML being behind the code, and it belongs in the file for the same reason it belongs on the page. """ profile = profile if isinstance(profile, dict) else {} return { 'profile': profile.get('profile'), 'title': profile.get('title'), 'member_kind': profile.get('member_kind'), 'panel_type': profile.get('panel_type'), 'basis': profile.get('basis') or 'none', 'unlisted': profile.get('unlisted') or 'ask', 'why': profile.get('why') or None, 'problem': profile.get('problem'), # No profile means the FULL master checklist under its own # numbering -- the compatibility contract, and a real answer. 'renumbered': bool(profile.get('profile')), 'off_board': list(off_board_ids or []), } def _panel_type_json(classification): """The classification, or an explicit unclassified answer. Never absent. `panel_type: null` is a REAL ANSWER here -- "nothing could tell this piece apart" -- and it travels with the sentence that says which signals said what. `used_for_applicability` is the other half nobody may infer: while `panel_types.detect_verified` is false the classification is reported and is NOT allowed to switch a check off, and a file that showed the type without saying so would read as though it had been. """ if not classification: return {'panel_type': None, 'basis': None, 'verified': False, 'used_for_applicability': False, 'why': None, 'candidates': [], 'signals': {}, 'facts': {}} verified = bool(classification.get('verified')) return { 'panel_type': classification.get('panel_type'), 'basis': classification.get('basis'), 'verified': verified, 'used_for_applicability': bool( verified and classification.get('panel_type')), 'why': classification.get('why'), 'candidates': list(classification.get('candidates') or []), 'signals': classification.get('signals') or {}, 'facts': classification.get('facts') or {}, } def _applicability_json(notes): """`ticketpage.apply_applicability`'s notes, named rather than positional. `declined` is the case that matters, and it is the mirror of muting's: a row that already FAILED keeps its FAIL, so the declaration did not take effect. A consumer reading only `checks.not_applicable` off the knowledge would believe the row was never asked; this says it was, and what it found. """ out = [] for row in (notes or []): num, name, reason, condition, since, by, layer, why = ( list(row) + [None] * 8)[:8] out.append({'item': num, 'name': name, 'reason': reason if reason else None, 'reason_missing': not reason, 'condition': condition, 'since': since, 'declared_by': by, 'layer': layer, 'layer_known': layer is not None, 'why': why, 'declined': bool(why)}) return out def _refusal_json(problems): """Declarations that were found and REFUSED, so their rows RAN. In the file for the same reason it is on the page: somebody wrote a rule believing a row is off. A refusal nobody records is that person never finding out it is on. """ return [{'item': num, 'problem': text} for (num, text) in (problems or [])] def _log_json(log): """`run_all`'s (check_name, state, note) tuples, named rather than positional. The disposition is the premise of every row's state, and a consumer reading `[..., "RAN-EMPTY", null]` off a bare array has to know the tuple order to use it.""" out = [] for row in (log or []): name, state, note = (list(row) + [None, None, None])[:3] out.append({'check': name, 'disposition': state, 'note': note}) return out def _muted_json(notes): """`apply_muting`'s (number, name, reason, why) notes. `declined` is the case that matters: a FAIL is never softened, so a mute on a failing row is REFUSED and the row keeps its own word. A consumer that read only `checks.muted` off the job file would believe the row was quiet; this says it was not. """ out = [] for row in (notes or []): num, name, reason, why = (list(row) + [None, None, None, None])[:4] out.append({'item': num, 'name': name, 'reason': reason if reason else None, 'reason_missing': not reason, 'why': why, 'declined': bool(why)}) return out # ------------------------------------------------------- every checklist row NO_ROW_REASON = ( 'the run produced no row for this checklist item at all. That is a ' 'defect in the tool, not a verdict on the piece -- read it as NOT RUN ' 'with an unknown reason, never as a pass') def checklist_rows_json(rows, unmuted_rows=None, log=None, muted_notes=None, applicability=None): """EVERY checklist row, with its state and its reason. Never a subset. THE ROWS COME FROM `ticketpage.CHECKLIST_ITEMS`, not from `rows`. That is the whole point and it is structural, not a promise: this walks the checklist table and emits one entry per number in it, then looks each number up in what the run produced. A row the run dropped -- for any reason, including one nobody has thought of yet -- lands with `reported_by_run: false` and `NO_ROW_REASON`, so a reader never has to infer a state from an absence. There is no list of row numbers in this file to fall out of step with the table. `rows` are the rows AS PRINTED (after `annotate_rows` and `apply_muting`) so the file and the page describe the same run. `unmuted_rows` are the same rows BEFORE muting, which is the only way to record what a muted row really reported as a state rather than as a sentence inside a note. A row in `rows` that is not in the checklist table is not dropped either: it comes back at the end with `in_checklist: false`. Nothing produces one today, and something silently swallowing it is exactly how it would stop being true. """ printed = {} for row in (rows or []): printed[row[0]] = row before = {} for row in (unmuted_rows or []): before[row[0]] = row muted_by_item = {} for note in (muted_notes or []): muted_by_item[note[0]] = note na_by_item = {} for note in (applicability or []): na_by_item[note[0]] = note log_by_name = {} for entry in (log or []): name, state, note = (list(entry) + [None, None, None])[:3] log_by_name[name] = (state, note) out = [] for item in TP.CHECKLIST_ITEMS: num, name, grade = item[0], item[1], item[2] owners, declared_reason, check_slot = item[3], item[4], item[5] checks = TP._check_names(check_slot) row = printed.get(num) if row is None: out.append(_row_json( num, name, grade, 'NOT RUN', NO_ROW_REASON, owners=owners, checks=checks, log_by_name=log_by_name, declared_reason=declared_reason, reported_by_run=False)) continue state = row[3] if len(row) > 3 else None note = row[4] if len(row) > 4 else None shown, _stable = TP.row_display(row) entry = _row_json(num, name, grade, state, note, owners=owners, checks=checks, log_by_name=log_by_name, declared_reason=declared_reason, reported_by_run=True, display_number=shown) if TP.state_key(state) == 'not_applicable': note_row = na_by_item.get(num) or (None,) * 7 _n, _name, reason, condition, since, by, why = ( list(note_row) + [None] * 7)[:7] entry['not_applicable'] = { 'condition': condition, 'reason': reason if reason else None, 'reason_missing': not reason, 'since': since, 'declared_by': by, # The check did not run: there is no verdict behind this row # and there is not meant to be. That is what separates it # from `muted`, which records what the check really said. 'check_ran': False, } if TP.state_key(state) == 'muted': was = before.get(num) was_state = was[3] if (was is not None and len(was) > 3) else None mute = muted_by_item.get(num) or (None, None, None, None) entry['muted'] = { # What the check REALLY reported. A grey row whose original # state survives only inside an English note is a state a # machine cannot read back, and muting is a volume control: # the answer is still there or it was never a mute. 'reported_state': was_state, 'reported_state_key': TP.state_key(was_state), 'reported_state_known': was_state is not None, 'reason': mute[2] if mute[2] else None, 'reason_missing': not mute[2], } # MUTING A ROW DOES NOT JUDGE IT. `MUTED` is listed as a judged # state because the check RUNS either way -- but a job may mute a # row that reported NOT RUN, and then the grey says "we asked for # this quietly" over a question nobody could answer. `judged` has # to follow what the row REALLY reported, or a consumer reading # "muted, judged: true" has been handed the silence-is-a-pass # reading through the one door this table left open. entry['judged'] = TP.state_is_judged(was_state) entry['reason'] = (None if entry['judged'] else (note or declared_reason)) if was_state is None: entry.setdefault('missing', {})['reported_state'] = ( 'this row was muted and the state it reported before ' 'muting was not recorded, so what the check actually ' 'said cannot be read back from this file') out.append(entry) known = set([i[0] for i in TP.CHECKLIST_ITEMS]) for num in sorted([n for n in printed if n not in known]): row = printed[num] extra = _row_json(num, row[1] if len(row) > 1 else None, row[2] if len(row) > 2 else None, row[3] if len(row) > 3 else None, row[4] if len(row) > 4 else None, owners=None, checks=(), log_by_name=log_by_name, declared_reason=None, reported_by_run=True) extra['in_checklist'] = False out.append(extra) return out def _row_json(num, name, grade, state, note, owners=None, checks=(), log_by_name=None, declared_reason=None, reported_by_run=True, display_number=None): """One checklist row as data. `reason` is filled ONLY where the row produced no verdict -- that is what requirement "a NOT RUN row carries its reason" means, and duplicating the finding message into it for a FAIL would blur the two. `note` is the row's note verbatim in every case, uncut, exactly as the page and `last-ticketqc.txt` print it. """ judged = TP.state_is_judged(state) row = { # THE MASTER ID. The stable identity, and what every mute, # applicability rule, ruling and test names. It never changes when a # plant renumbers. 'item': num, 'master_id': num, 'name': name, 'grade': grade, 'state': state, 'state_key': TP.state_key(state), 'judged': judged, 'in_checklist': True, 'reported_by_run': reported_by_run, 'note': note, 'reason': None if judged else (note or declared_reason), # WHAT THE PLANT'S BOARD CALLED IT. Equal to the master id when no # profile renumbered this row, and always present so a consumer # never has to work out which of the two it is holding. 'display_number': display_number or num, 'renumbered': bool(display_number and display_number != num), 'owner_codes': list(owners or []), 'checks': [ {'check': cn, 'disposition': (log_by_name or {}).get(cn, (None, None))[0], 'note': (log_by_name or {}).get(cn, (None, None))[1]} for cn in (checks or [])], } if not judged and not row['reason']: row.setdefault('missing', {})['reason'] = ( 'this row produced no verdict and recorded no reason. Read it as ' 'not judged; it is not a pass') return row def _json_row(f, default_source): """Accepts a Finding or one of this module's rows -- both carry the same three fields, and duck-typing them keeps one shape in the JSON.""" if isinstance(f, dict): row = {'severity': f.get('severity'), 'code': f.get('code'), 'message': f.get('message'), 'source': f.get('source') or default_source} extra = f.get('model_read') else: row = {'severity': f.severity, 'code': f.code, 'message': f.message, 'source': getattr(f, 'source', None) or default_source} extra = getattr(f, 'model_read', None) row['computed'] = (row['source'] == 'computed') if extra: row['model_read'] = extra return row def vision_dir(root_dir): """The vision working folder: requests, responses, logs, rendered views. Just `vision` under the QC data folder -- `designspec.root_dir` already OWNS the layout and in the beside-the-model one it ends in `_ticketqc` itself, so joining that segment again put every file in `/_ticketqc/_ticketqc/vision`. (Fixed 2026-08-30. Nothing reads this folder back across runs -- requests and responses are written and read inside one press -- so an old doubled folder is stale scratch, not lost data, and is left where it is rather than migrated.) """ return os.path.join(root_dir or os.path.expanduser('~'), 'vision') def write_run_json(root_dir, document, name='qc-run.json'): """The run's findings, both kinds, beside the model's QC data. Written on EVERY press, next to `last-ticketqc.txt`, whether or not the vision escalation ran -- `model_read` is simply an empty list when it did not. The page promises the muted section's findings "still reach the machine-readable JSON" on every run with mutes, and vision defaults off, so a file that appeared only on an escalated run made that sentence describe something that was not there. Whether a model was consulted is still readable at a glance: `model_read` is empty when it was not. """ folder = root_dir or os.path.expanduser('~') if not os.path.isdir(folder): os.makedirs(folder) path = os.path.join(folder, name) text = json.dumps(document, indent=2, sort_keys=True) if not isinstance(text, type(u'')): text = text.decode('utf-8') with io.open(path, 'w', encoding='utf-8') as fh: fh.write(text) return path def load_run_json(path): """The run document at `path`, or an EMPTY one saying why. Copied from `reinforcecore.load()` and `ticketqc.load_evidence`, down to the shape of the failure: a file whose `schema` is not `RUN_SCHEMA` is discarded WHOLE. Not migrated, not partially adopted, and above all not read for the keys that happen to still parse. Half a run file is the worst possible outcome here and it is worth spelling out why. The keys a `qc-run/0` file (there was no such word -- the old file carried no schema at all) still parses under are `findings` and `model_read`. A reader that adopted those two and skipped `tickets` would be reading exactly the file this schema exists to replace: a list of findings with no record of which questions were asked, which reads as a clean run. So `tickets` comes back EMPTY on any failure, and a caller that ignores `_load_error` judges nothing rather than judging a run it cannot see the shape of. """ def _empty(why): return {'schema': RUN_SCHEMA, 'written_at': None, 'written_by': None, 'build': None, 'doc_key': None, 'model_stem': None, 'knowledge_sources': None, 'ticket_count': 0, 'tickets': [], 'findings': [], 'model_read': [], 'model_read_disclaimer': DISCLAIMER, 'missing': {}, '_load_error': why} if not path or not os.path.isfile(path): return _empty('no run file at {0}'.format(path or '(no path)')) try: with io.open(path, 'r', encoding='utf-8') as fh: document = json.loads(fh.read()) except Exception as exc: # pylint: disable=W0703 return _empty('{0}: {1}'.format(type(exc).__name__, str(exc)[:120])) if not isinstance(document, dict) or document.get('schema') != RUN_SCHEMA: return _empty('schema {0}, expected {1}'.format( (document or {}).get('schema') if isinstance(document, dict) else '?', RUN_SCHEMA)) document.setdefault('tickets', []) return document # --------------------------------------------------------- the report section def report_block(model_read): """Markdown for the model-read rows: their own heading, their own table, their own column naming what read them. Never merged into the checklist table or the issue tally. A reader must be able to tell at a glance which findings were computed and which were read off a picture, and the surest way is that they are never in the same place. """ if not model_read: return None, [] head = ('#### Read by a model, not computed  —  not a ' 'verdict\n\n> ' + DISCLAIMER) rows = [] for entry in model_read: label, f = entry if isinstance(entry, tuple) else ('-', entry) row = _json_row(f, 'model-read') info = row.get('model_read') or {} rows.append([label, row['code'], info.get('view') or '-', info.get('answer') or info.get('status') or '-', 'a model looked at a picture', row['message']]) return head, rows REPORT_COLUMNS = ['ticket', 'code', 'view', 'the model said', 'read by', 'what that means'] # ------------------------------------------------- the checklist row annotation # # Additive on purpose: `ticketpage.checklist_states` is untouched. It decides # the computed status of every row and knows nothing about `AI-*` codes, # which own no row in CHECKLIST_ITEMS and therefore cannot flip one. This # takes its output and marks the rows an escalation touched, so the provenance # reaches the checklist table without a model read ever being able to set a # row's status. def annotate_rows(rows, model_read): """checklist rows -> the same rows, with escalated ones marked. The status is NEVER changed -- a row a model looked at stays NOT RUN or FOR ENGINEER, which is what section 8bis requires. Only the note grows a marker, and the marker says where to read the rest. """ by_item = {} for f in (model_read or []): code = f.get('code') if isinstance(f, dict) else f.code for row in ESCALATIONS: if row['code'] == code: by_item.setdefault(row['item'], []).append(f) out = [] for r in rows: num, name, grade, status, note = r if num in by_item: mark = ('[a model also looked -- see "Read by a model"; that is ' 'not a verdict and this row did not run]') note = (note + ' ' + mark) if note else mark out.append((num, name, grade, status, note)) return out # --------------------------------------------------------------- the driver def escalation_items(t, know, root_dir, options=None): """What this ticket would escalate, and why it will not. Returns (items, blocked) -- `items` are request rows ready to render and send; `blocked` are WARN findings for the escalations that stopped before the model, each naming its reason. Every gate below is a REPORTED reason, never a silent skip: * the run option is off * the plant has not named the views this check needs * no such view is on this ticket * the view prints too small to render legibly Rendering itself is left to the caller, which is the only part that needs Revit. """ on, cap = run_settings(root_dir, options) items, blocked = [], [] for row in ESCALATIONS: if not on: blocked.append(not_run_row(row['code'], row['what'], vision_off_reason())) continue names = _plant_views(row, know) if not names: blocked.append(not_run_row( row['code'], row['what'], 'no plant has recorded which views carry this ({0} is empty), ' 'so there is no view to render -- name them in the knowledge ' 'file'.format(row['views_key']))) continue found = _views_matching(t, names) if not found: blocked.append(not_run_row( row['code'], row['what'], 'this ticket places no view named in {0}, so there was ' 'nothing to look at'.format(row['views_key']))) continue for (view_name, viewport) in found: w, h = viewport_paper_size_in(viewport) plan = render_plan(w, h) if not plan['ok']: blocked.append(not_run_row(row['code'], row['what'], plan['reason'], view=view_name)) continue items.append({ 'id': '{0}|{1}|{2}'.format(t.get('mark'), row['item'], view_name), 'question_id': row['question_id'], 'code': row['code'], 'item': row['item'], 'what': row['what'], 'view': view_name, 'view_kind': 'reinforcing view off a shop ticket', 'viewport': viewport, 'pixels': plan['pixels'], 'effective_dpi': round(plan['effective_dpi'], 1), }) if cap and len(items) > cap: for item in items[cap:]: blocked.append(not_run_row( item['code'], item['what'], 'this run\'s vision call budget ({0}, run option {1}) was ' 'already spent'.format(cap, OPTION_MAX), view=item['view'])) items = items[:cap] return items, blocked def read_for_ticket(t, know, root_dir, options=None, timeout_ms=180000): """The whole escalation for one ticket, as `Finding` objects. Parallel in shape to `run_all`'s findings and different in every way that matters: `source` says 'model-read', the codes are `AI-*`, the severity is always WARN, and they arrive in their own list. A caller that merges the two lists by accident still cannot lose the difference. """ return to_findings(read_rows_for_ticket(t, know, root_dir, options, timeout_ms)) def read_rows_for_ticket(t, know, root_dir, options=None, timeout_ms=180000): """The escalation as plain rows. `read_for_ticket` wraps this in Findings. Render -> request file -> CPython -> response file -> rows. Never raises, never returns nothing: every path out of here produces one row per escalation, because a silent one is read as a pass. The CPython half is reached exactly the way the handling check reaches panellift: a one-line cmd wrapper redirecting both streams to files. Two pipes deadlock the moment the child outwrites a buffer -- see `liftexport.run` for the version of that lesson that was paid for. """ from System.Diagnostics import Process, ProcessStartInfo from vibe import liftexport items, blocked = escalation_items(t, know, root_dir, options) if not items: return blocked work = vision_dir(root_dir) try: if not os.path.isdir(work): os.makedirs(work) except Exception: pass python_exe = liftexport.find_python(know) if not python_exe: return blocked + [not_run_row( i['code'], i['what'], 'no CPython 3 was found to make the model call with -- set ' 'handling.python_exe or WILLTOOLS_PYTHON', view=i['view']) for i in items] sendable, subjects = [], {} for item in items: subjects[item['id']] = {'what': item['what'], 'view': item['view']} try: png = render_view_png(t['doc'], item['viewport'].ViewId, work, item['pixels']) except Exception as ex: blocked.append(not_run_row( item['code'], item['what'], 'the view could not be rendered to an image ({0}), so there ' 'was no picture to look at'.format(type(ex).__name__), view=item['view'])) continue sendable.append({'id': item['id'], 'question_id': item['question_id'], 'code': item['code'], 'image': png, 'view': item['view'], 'view_kind': item['view_kind'], 'effective_dpi': item['effective_dpi']}) if not sendable: return blocked stamp = str(int(time.time() * 1000)) req_path = os.path.join(work, 'request_' + stamp + '.json') res_path = os.path.join(work, 'response_' + stamp + '.json') runner = os.path.join(work, 'run_' + stamp + '.cmd') log = os.path.join(work, 'run_' + stamp + '.log') _, cap = run_settings(root_dir, options) write_request(req_path, build_request(sendable, max_calls=cap)) script = os.path.join(os.path.dirname(os.path.dirname( os.path.dirname(os.path.dirname(os.path.abspath(__file__))))), 'tools', 'vision_read.py') with io.open(runner, 'w', encoding='utf-8') as fh: fh.write(u'@echo off\r\n') fh.write(u'chcp 65001 >nul\r\n') fh.write(u'set "PYTHONIOENCODING=utf-8"\r\n') # The credential is NEVER written here. It is read from the # environment by the SDK in the child, so it never appears in this # file, in the request, in the response, or in the log. fh.write(u'"{0}" "{1}" "{2}" "{3}" > "{4}" 2>&1\r\n' .format(python_exe, script, req_path, res_path, log)) fh.write(u'exit /b %ERRORLEVEL%\r\n') info = ProcessStartInfo() info.FileName = 'cmd.exe' info.Arguments = '/c "{0}"'.format(runner) info.UseShellExecute = False info.CreateNoWindow = True try: process = Process.Start(info) waited = 0 while not process.WaitForExit(250): waited += 250 if waited >= timeout_ms: try: process.Kill() except Exception: pass return blocked + [not_run_row( i['code'], subjects[i['id']]['what'], 'the model call did not return within {0} seconds' .format(int(timeout_ms / 1000)), view=i['view']) for i in sendable] except Exception as ex: return blocked + [not_run_row( i['code'], subjects[i['id']]['what'], 'the model call could not be started ({0})'.format( type(ex).__name__), view=i['view']) for i in sendable] response = read_response(res_path) if not response: return blocked + [not_run_row( i['code'], subjects[i['id']]['what'], 'the model call wrote no answer -- see ' + os.path.basename(log), view=i['view']) for i in sendable] return blocked + rows_from_response(response, subjects) def _views_matching(t, names): """(view name, viewport) for every placed view whose name starts with one of the plant's declared names. Case-insensitive; the plant writes the names, not the code.""" wanted = [str(n).strip().upper() for n in names if str(n).strip()] viewports = t.get('viewports') if viewports is None and t.get('doc') is not None and t.get('sheets'): try: viewports = collect_viewports(t['doc'], t['sheets']) t['viewports'] = viewports except Exception: viewports = [] out = [] for vp in (viewports or []): try: name = vp['name'] except Exception: continue upper = (name or u'').strip().upper() for w in wanted: if upper.startswith(w): out.append((name, vp.get('viewport'))) break return out