open/jobbsok
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity Actions
jobbsok/scripts/sak_status.py
Kjell Tore Guttormsen 12125c1932 feat(m2): add deterministic sak_status cli with golden output 
--workspace is required with no environment fall-back: this process runs
unsandboxed, and a status run that quietly picked a workspace is one nobody
can audit. --check exits 1 on a stale cache, distinct from 2 for a failed run.

Byte-stability is measured, not asserted: the CLI is run twice under two
PYTHONHASHSEED values and the bytes compared, in both output formats. The
write boundary gets two digest tests rather than one -- without --oppdater
every byte stays put, with it exactly the diverging case's four cached keys
move and logg.jsonl, the body and every other file are byte-identical.

Co-Authored-By: Claude <claude-opus-5>
2026-09-05 21:40:32 +02:00

521 lines
20 KiB
Python
Raw Blame History

"""The deterministic status machine behind every case (plan Steps 17-20).
Status is derived from `logg.jsonl` and from nothing else. `sak.md`
frontmatter is a cache of that derivation, and where the two disagree the log
wins and the divergence is reported rather than quietly repaired (risk H3).
A cache nothing refreshes is only permanent divergence with better manners,
so this module also owns the write-back: :func:`oppdater` rewrites exactly
four frontmatter keys -- `status`, `ventende_part`, `sist_aktivitet`,
`neste_frist` -- and touches no other key and no body line. It is the ONLY
writer of those four keys, and every other path through this module is a pure
reader.
Three decisions this step had to take, stated here rather than left implicit.
**Two rows of the transition table are not events.** The plan writes "Decision
`ja`" and "Operator close" without the word *event*, and neither is in the
build-brief 5.3 enum, which is closed at eleven and stays closed. They reach
the case log as the decision-log records build-brief 5.7 already mirrors
there: a `type: beslutning` line carrying `beslutning: ja`, and a
`type: utfall` line carrying `utfall: avsluttet`. So the case log is a stream
of *triggers* with two spellings -- `hendelse` for the eleven events, `type`
for the two mirrored records -- and :func:`trigger_of` is the single place
that decides which. A `beslutning: nei` in a case log is not a trigger: a no
creates no case and advances none.
**`neste_frist` is the date this case goes silent.** Nothing in the brief
derives it, and `--oppdater` is required to write it from derived truth, so it
has to mean something the machine can compute. It is the earliest threshold
date among the silence rules that apply to the case right now, and `null`
where no rule applies.
**A derivation of `None` does not overwrite the cache.** `neste_frist` in
`vurderer` and `soker` is the application deadline, which is the operator's
own data and which no machine here can recompute. A refresh that emptied the
one field the operator typed would be a defect wearing a refresh's clothes,
so :func:`oppdater` writes a key only when the derivation has a value for it,
and :func:`divergenser` likewise compares only those. This is a deliberate
deviation from a literal reading of "rewrites the four keys"; it is the only
one in this module.
Determinism is a property of the interface. The clock is injected as a date,
never read off the wall, and nothing that reaches stdout is iterated out of a
dict or a set -- two runs are byte-identical, including under different hash
seeds.
"""
import datetime
import os
from jobbsok_lib import frontmatter as frontmatter_lib
from jobbsok_lib import jsonl, paths
#: The closed event vocabulary of build-brief 5.3. An event outside it is an
#: error naming this whole set, never a silent skip.
HENDELSER = (
"opprettet",
"soknad_sendt",
"bekreftelse_mottatt",
"henvendelse_mottatt",
"svar_sendt",
"intervju_avtalt",
"intervju_gjennomfort",
"tilbud_mottatt",
"avslag",
"trukket",
"stille",
)
#: The decision mirrored into the case log, and the operator's close. Not
#: events; see the module docstring.
BESLUTNING_JA = "beslutning_ja"
AVSLUTTET = "avsluttet"
#: Everything the table can be keyed by.
TRIGGERE = HENDELSER + (BESLUTNING_JA, AVSLUTTET)
#: build-brief 6: the main chain, then the two terminal side-states.
TILSTANDER = (
"vurderer", "soker", "sendt", "dialog", "intervju", "tilbud", "avsluttet",
"avslag", "trukket",
)
#: Nothing leaves these. An event after one of them is an error, not a
#: reopening -- a case that starts again is a new case with a new sak-id.
TERMINALE = ("avsluttet", "avslag", "trukket")
#: The state and waiting party each trigger lands on. `stille` maps to None:
#: it is operator-appended, carries no transition, and this module never
#: appends it (there is no autonomous scheduling).
OVERGANGER = {
"opprettet": ("vurderer", "meg"),
BESLUTNING_JA: ("soker", "meg"),
"soknad_sendt": ("sendt", "dem"),
"bekreftelse_mottatt": ("sendt", "dem"),
"henvendelse_mottatt": ("dialog", "meg"),
"svar_sendt": ("dialog", "dem"),
"intervju_avtalt": ("intervju", "ingen"),
"intervju_gjennomfort": ("intervju", "dem"),
"tilbud_mottatt": ("tilbud", "meg"),
AVSLUTTET: ("avsluttet", "ingen"),
"avslag": ("avslag", "ingen"),
"trukket": ("trukket", "ingen"),
"stille": None,
}
#: Which triggers are legal from which state. A trigger whose target is the
#: state itself is listed, so a duplicate event is idempotent rather than an
#: error -- an operator who logs the same send twice made a bookkeeping slip,
#: not a state error. `avslag`, `trukket`, `avsluttet` and `stille` are legal
#: from every non-terminal state: a case can be rejected, withdrawn, closed or
#: noted as quiet at any point.
_ALLTID = ("avslag", "trukket", AVSLUTTET, "stille")
LOVLIGE = {
"vurderer": ("opprettet", BESLUTNING_JA) + _ALLTID,
"soker": (BESLUTNING_JA, "soknad_sendt") + _ALLTID,
"sendt": (
"soknad_sendt", "bekreftelse_mottatt", "henvendelse_mottatt",
"intervju_avtalt", "tilbud_mottatt",
) + _ALLTID,
"dialog": (
"henvendelse_mottatt", "svar_sendt", "intervju_avtalt", "tilbud_mottatt",
) + _ALLTID,
"intervju": (
"intervju_avtalt", "intervju_gjennomfort", "henvendelse_mottatt",
"tilbud_mottatt",
) + _ALLTID,
"tilbud": ("tilbud_mottatt",) + _ALLTID,
"avsluttet": (),
"avslag": (),
"trukket": (),
}
#: Events that mean the other party actually made contact. An automated
#: receipt is not contact, which is why `bekreftelse_mottatt` is absent: the
#: plan says it must not reset the silence clock, and leaving it out here is
#: what makes that true rather than commented.
INNKOMMENDE = ("henvendelse_mottatt", "intervju_avtalt", "tilbud_mottatt")
#: Flag names in the order they are emitted. Fixed, so output ordering never
#: depends on which rule happened to fire first.
FLAGG = ("sendt_14", "dialog_7", "intervju_10")
#: build-brief 6's three silence thresholds, in days.
TERSKLER = {"sendt_14": 14, "dialog_7": 7, "intervju_10": 10}
#: The frontmatter keys this module caches, and the only ones it ever writes.
CACHE_KEYS = ("status", "ventende_part", "sist_aktivitet", "neste_frist")
#: The one non-value a frontmatter scalar can carry. `null` reads back as the
#: string "null", so the mapping is done here rather than in the parser, which
#: must keep returning the operator's bytes untouched.
NULL = "null"
class StatusError(Exception):
"""A refusal from the status machine, naming what would have been legal."""
def trigger_of(record):
"""Map a case-log record to a trigger name, or None when it carries none.
Raises :class:`StatusError` for a record that names an event outside the
closed enum, and for one this module cannot read at all. Returning None is
reserved for records that are legitimately not transitions -- a
`beslutning: nei`, a `korrigering`, an `utfall` that is not the close.
"""
if not isinstance(record, dict):
raise StatusError("a case-log record must be an object, got %s"
% type(record).__name__)
if "hendelse" in record:
hendelse = record["hendelse"]
if hendelse not in HENDELSER:
raise StatusError(
"%r is not a case event. The vocabulary is closed: %s"
% (hendelse, ", ".join(HENDELSER))
)
return hendelse
kind = record.get("type")
if kind == "beslutning":
return BESLUTNING_JA if record.get("beslutning") == "ja" else None
if kind == "utfall":
return AVSLUTTET if record.get("utfall") == AVSLUTTET else None
if kind == "korrigering":
return None
raise StatusError(
"a case-log record must carry 'hendelse' from the 5.3 enum or 'type' "
"from %r; got %r" % (list(jsonl.TYPES), sorted(record))
)
def avgjor(records, today):
"""Derive status, waiting party, dates and silence flags from a case log.
``today`` is injected, always: a machine that only behaves deterministically
because a test patched its clock is a machine its callers cannot make
behave deterministically at all.
"""
dagen = _dato(today)
ordered = jsonl.by_time(records)
status, ventende = "vurderer", "meg"
sist_aktivitet = None
sist_innkommende = None
sendt_fra = None
intervju_holdt = None
utfall_etter_intervju = False
for record in ordered:
trigger = trigger_of(record)
if trigger is None:
continue
if status in TERMINALE:
raise StatusError(
"%r comes after the terminal state %r; a case that starts "
"again is a new case with a new sak-id" % (trigger, status)
)
if trigger not in LOVLIGE[status]:
raise StatusError(
"%r is not legal from %r. Legal here: %s"
% (trigger, status, ", ".join(sorted(LOVLIGE[status])))
)
maal = OVERGANGER[trigger]
if maal is not None:
status, ventende = maal
naa = jsonl.instant(record).date()
sist_aktivitet = naa if sist_aktivitet is None or naa > sist_aktivitet else sist_aktivitet
if trigger in INNKOMMENDE:
sist_innkommende = naa
if trigger == "soknad_sendt":
sendt_fra = naa
if trigger == "intervju_gjennomfort":
intervju_holdt = naa
utfall_etter_intervju = False
if intervju_holdt is not None and trigger in ("tilbud_mottatt",) + TERMINALE:
utfall_etter_intervju = True
frister = _frister(status, ventende, sist_innkommende, sendt_fra,
intervju_holdt, utfall_etter_intervju)
flagg = [navn for navn in FLAGG if navn in frister and frister[navn] <= dagen]
neste_frist = min(frister.values()) if frister else None
return {
"status": status,
"ventende_part": ventende,
"sist_aktivitet": _iso(sist_aktivitet),
"neste_frist": _iso(neste_frist),
"flagg": flagg,
"innkommende": _innkommende(sist_innkommende),
}
def _frister(status, ventende, sist_innkommende, sendt_fra, intervju_holdt,
utfall_etter_intervju):
"""The date each applicable silence rule trips, keyed by flag name.
Terminal states are never in here, and `vurderer` and `soker` never are
either: nobody is waiting on the other party yet, so there is nothing for
silence to mean.
"""
frister = {}
if status in TERMINALE:
return frister
if status == "sendt":
basis = sist_innkommende or sendt_fra
if basis is not None:
frister["sendt_14"] = basis + datetime.timedelta(days=TERSKLER["sendt_14"])
if status == "dialog" and ventende == "dem" and sist_innkommende is not None:
frister["dialog_7"] = sist_innkommende + datetime.timedelta(days=TERSKLER["dialog_7"])
if intervju_holdt is not None and not utfall_etter_intervju:
frister["intervju_10"] = intervju_holdt + datetime.timedelta(days=TERSKLER["intervju_10"])
return frister
def _innkommende(sist):
"""What is known about inbound contact -- and what is only *not recorded*.
With the mail server deferred, "no inbound" is a fact about this log and
not about the world (risk H9). Saying so is the difference between a daily
view worth reading and one that flags every case as gone quiet.
"""
if sist is not None:
return {"registrert": True, "sist": _iso(sist), "merknad": None}
return {
"registrert": False,
"sist": None,
"merknad": (
"ingen innkommende hendelse er registrert. Med e-postserveren "
"utsatt betyr det ikke at ingen finnes."
),
}
def alle_saker(root):
"""Every sak-id under ``root/saker``, sorted. Sorted, not os.listdir order."""
saker = paths.safe_join(root, "saker")
if not os.path.isdir(saker):
return []
funnet = []
for navn in sorted(os.listdir(saker)):
if not os.path.isdir(os.path.join(saker, navn)):
continue
funnet.append(paths.validate_sak_id(navn))
return funnet
def les_sak(root, sak_id):
"""Return ``(metadata, body, records)`` for one case. Pure reader."""
paths.validate_sak_id(sak_id)
katalog = paths.safe_join(root, "saker", sak_id)
with open(os.path.join(katalog, "sak.md"), "r", encoding="utf-8") as handle:
meta, body = frontmatter_lib.parse(handle.read())
return meta, body, jsonl.read_lines(os.path.join(katalog, "logg.jsonl"))
def status_for_sak(root, sak_id, today):
"""The derived truth for one case, plus how the cache disagrees with it."""
meta, _body, records = les_sak(root, sak_id)
resultat = avgjor(records, today)
resultat["sak_id"] = sak_id
resultat["arbeidsgiver"] = meta.get("arbeidsgiver")
resultat["rolle"] = meta.get("rolle")
resultat["divergens"] = _divergens(meta, resultat)
return resultat
def rapport(root, today, sak=None):
"""The whole workspace as one ordered structure. Writes nothing."""
saker = [paths.validate_sak_id(sak)] if sak else alle_saker(root)
return {
"dato": _iso(_dato(today)),
"saker": [status_for_sak(root, sak_id, today) for sak_id in saker],
}
def divergenser(root, today, sak=None):
"""Every case whose cached frontmatter disagrees with its log."""
ut = []
for resultat in rapport(root, today, sak)["saker"]:
if resultat["divergens"]:
ut.append({"sak_id": resultat["sak_id"], "divergens": resultat["divergens"]})
return ut
def oppdater(root, sak_id, resultat):
"""Rewrite the four cached keys in ``sak.md``. The only writer here.
Returns the keys that actually changed. A key whose derivation is None is
left alone -- see the module docstring.
"""
sti = paths.safe_join(root, "saker", sak_id, "sak.md")
with open(sti, "r", encoding="utf-8") as handle:
meta, body = frontmatter_lib.parse(handle.read())
endret = []
for nokkel in CACHE_KEYS:
utledet = resultat.get(nokkel)
if utledet is None:
continue
if _hurtigbuffer(meta, nokkel) != utledet:
meta[nokkel] = utledet
endret.append(nokkel)
if endret:
with open(sti, "w", encoding="utf-8") as handle:
handle.write(frontmatter_lib.render(meta, body))
return endret
def _divergens(meta, resultat):
avvik = []
for nokkel in CACHE_KEYS:
utledet = resultat.get(nokkel)
if utledet is None:
continue
hurtigbuffer = _hurtigbuffer(meta, nokkel)
if hurtigbuffer != utledet:
avvik.append({
"nokkel": nokkel,
"hurtigbuffer": hurtigbuffer,
"utledet": utledet,
})
return avvik
def _hurtigbuffer(meta, nokkel):
verdi = meta.get(nokkel)
return None if verdi == NULL else verdi
def _dato(value):
if isinstance(value, datetime.date) and not isinstance(value, datetime.datetime):
return value
try:
return datetime.date.fromisoformat(str(value))
except ValueError:
raise StatusError("--today must be YYYY-MM-DD, got %r" % (value,))
def _iso(value):
return None if value is None else value.isoformat()
# ---------------------------------------------------------------------------
# Command line (plan Step 20)
#
# Output ordering is fully determined: cases come out sorted by sak-id, keys
# in JSON are sorted, and flags come out in FLAGG order. Nothing here iterates
# a dict or a set into stdout, which is what makes two runs byte-identical
# under different hash seeds rather than only usually.
# ---------------------------------------------------------------------------
#: Exit codes. 1 is reserved for `--check` finding divergence, so a caller can
#: tell "the cache is stale" apart from "the run failed".
EXIT_OK = 0
EXIT_DIVERGENS = 1
EXIT_FEIL = 2
def build_parser():
import argparse
parser = argparse.ArgumentParser(
prog="sak_status.py",
description=(
"Utled status, ventende part og stillhets-flagg for hver sak i "
"arbeidsomraadet. Leser logg.jsonl; sak.md er en hurtigbuffer."
),
)
# Required, and deliberately without the environment fall-backs
# paths.workspace_root offers: this process runs unsandboxed, and a
# status run that quietly picked a workspace would be a status run
# nobody could audit.
parser.add_argument("--workspace", required=True,
help="arbeidsomraadets rot (ingen underforstaatt standard)")
parser.add_argument("--today", default=None, metavar="YYYY-MM-DD",
help="klokka som skal brukes; injisert, aldri veggklokka")
parser.add_argument("--sak", default=None, metavar="SAK-ID",
help="begrens til en enkelt sak")
parser.add_argument("--format", dest="format", default="tekst",
choices=("tekst", "json"))
parser.add_argument("--check", action="store_true",
help="avslutt med kode 1 hvis en hurtigbuffer avviker")
parser.add_argument("--oppdater", action="store_true",
help="skriv de fire hurtigbuffer-noklene tilbake til sak.md")
return parser
def _skriv(strom, tekst):
strom.write(tekst)
strom.flush()
def _som_json(data):
import json as _json
return _json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
def _som_tekst(data):
linjer = ["sak-status %s (%d saker)" % (data["dato"], len(data["saker"])), ""]
for sak in data["saker"]:
linjer.append(sak["sak_id"])
linjer.append(" status: %-9s ventende: %-6s sist aktivitet: %s"
% (sak["status"], sak["ventende_part"],
sak["sist_aktivitet"] or "-"))
linjer.append(" neste frist: %s" % (sak["neste_frist"] or "-"))
if sak["flagg"]:
linjer.append(" flagg: %s" % ", ".join(sak["flagg"]))
if not sak["innkommende"]["registrert"]:
linjer.append(" innkommende: %s" % sak["innkommende"]["merknad"])
for avvik in sak["divergens"]:
linjer.append(" avvik: %s hurtigbuffer=%r utledet=%r"
% (avvik["nokkel"], avvik["hurtigbuffer"], avvik["utledet"]))
linjer.append("")
return "\n".join(linjer)
def main(argv=None, stdout=None, stderr=None):
import sys as _sys
argv = _sys.argv[1:] if argv is None else argv
stdout = _sys.stdout if stdout is None else stdout
stderr = _sys.stderr if stderr is None else stderr
args = build_parser().parse_args(argv)
today = args.today or datetime.date.today().isoformat()
try:
root = paths.workspace_root(args.workspace)
if args.sak is not None:
paths.validate_sak_id(args.sak)
data = rapport(root, today, args.sak)
if args.oppdater:
for sak in data["saker"]:
oppdater(root, sak["sak_id"], sak)
data = rapport(root, today, args.sak)
except (StatusError, paths.WorkspaceError, jsonl.JsonlError,
frontmatter_lib.FrontmatterError, OSError) as feil:
_skriv(stderr, "sak_status: %s\n" % feil)
return EXIT_FEIL
_skriv(stdout, _som_json(data) if args.format == "json" else _som_tekst(data))
if args.check:
avvikende = [sak for sak in data["saker"] if sak["divergens"]]
if avvikende:
_skriv(stdout, "\n%d sak(er) med utdatert hurtigbuffer. Kjor --oppdater.\n"
% len(avvikende))
return EXIT_DIVERGENS
return EXIT_OK
if __name__ == "__main__":
import sys as _sys
_sys.exit(main())
Reference in a new issue View git blame Copy permalink