open/jobbsok
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity Actions

feat(m2): add append-only decision log and beslutning skill

Browse source
This commit is contained in:
Kjell Tore Guttormsen 2026-09-05 22:00:13 +02:00
commit 06b45e9168
4 changed files with 761 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

446
scripts/beslutninger.py Normal file
View file

@ -0,0 +1,446 @@
"""The append-only sourcing-decision store from build-brief 5.4 (plan Step 22).
One line per decision the operator took about one listing. The store is the
input M6's learning loop reads, and it is append-only, which has a consequence
worth stating before the first line is written: **it cannot be migrated.**
There is no pass that can go back and add a field, fix a spelling or widen an
enum. So every rule this module will ever enforce is enforced from the first
append, and `type` and `skjema` are on every line (risk H10).
A correction is a new line, never an edit. `korriger` appends a `korrigering`
record naming the decision it restates; `jobbsok_lib.jsonl.fold` applies it on
read. Both lines stay on disk forever -- what the operator first thought is
part of the record, and a store that erased it would be a worse learning
signal than one that keeps both.
Three things this module deliberately does **not** do:
**It does not recompute `vekt_hash`.** The hash fingerprints the weight vector
the score was computed under, and computing it here would make the decision log
depend on `vurdering`, whose criteria are its own. What is checked is the shape
-- a `sha256:` string -- and that `delscore` and `vekter` describe the same set
of criteria. Which criteria those are is scoring's contract, not this store's,
and duplicating the list here would be a second copy free to drift from the
first.
**It does not score, and it does not decide.** `score_da` is handed in, and the
name says why it exists: the score as it stood *before* the human decided.
Build-brief 5.4 calls that the learning signal -- a high score paired with a
`nei` is the divergence 5.7 mines -- and a score recomputed at write time would
be the machine grading its own homework.
**It does not create the case.** On `ja` the `sak` skill does that, from the
sak-id :func:`sak_id_for` derives. This module offers
:func:`speil_i_sakslogg`, because the mirrored decision is the one record that
takes a case from `vurderer` to `soker`, and a case log missing it is a case
that never moves.
"""
from jobbsok_lib import jsonl, paths
#: The decision log, relative to the workspace root.
FILNAVN = "beslutninger.jsonl"
#: build-brief 5.4's closed reason enum, in the order the brief lists it.
ARSAKER = (
"lonn", "geografi", "arbeidsform", "fagomrade", "oppgavetype", "teknologi",
"senioritetsniva", "selskapstype", "arbeidsgiverrykte", "tidspunkt",
"konkurranse", "annet",
)
#: One to three of them, per 5.4. Three is a decision with nuance; a fourth is
#: a decision nobody can act on.
MIN_ARSAKER = 1
MAKS_ARSAKER = 3
#: 5.4's cap on the note. It keeps the field a label rather than a second body.
MAKS_NOTAT_ORD = 15
DECISIONS = ("ja", "nei")
#: The layout every decision line has, in the order the fixture corpus under
#: `tests/fixtures/beslutninger/` already writes it. That corpus is the fasit:
#: the store cannot be migrated, so a writer that laid the same fields out
#: differently would fork the format on line one.
FELTREKKEFOLGE = (
"type", "skjema", "id", "dato", "kilde", "url", "tittel", "arbeidsgiver",
"beslutning", "arsak", "notat", "score_da", "delscore", "vekter",
"vekt_hash", "korrigerer",
)
#: A correction's own fields, then whichever decision fields it restates.
KORRIGERING_HODE = ("type", "skjema", "id", "dato", "korrigerer")
#: What a correction is allowed to restate. Not `id`, not `dato`, not
#: `score_da`: the score as it stood when the human decided is a measurement,
#: and a measurement is not corrected by deciding differently later.
KORRIGERBARE = ("kilde", "url", "tittel", "arbeidsgiver", "beslutning", "arsak", "notat")
class BeslutningError(Exception):
"""A refusal from this module, naming the field it refused on."""
def sti(root):
"""The decision log's path under ``root``, resolved through safe_join."""
return paths.safe_join(root, FILNAVN)
def les(root):
"""Every line on disk, corrections and outcomes included. Pure reader."""
return jsonl.read_lines(sti(root))
def stromme(root):
"""The decision stream with corrections applied and outcomes removed."""
return jsonl.fold(les(root))
def legg_til(root, felt):
"""Validate ``felt`` and append it as exactly one decision line.
Validation runs before the file is opened, so a rejected decision leaves
no trace -- not a partial line, not an empty file.
"""
record = _bygg_beslutning(felt)
_krev_ubrukt_id(root, record["id"])
return jsonl.append_line(sti(root), record)
def korriger(root, felt):
"""Append a correction naming a decision that is already in the log.
The reference is checked here rather than left to read time: `jsonl.fold`
raises on a correction pointing at nothing, and a log that cannot be folded
is a log that cannot be read at all.
"""
record = _bygg_korrigering(felt)
kjente = {rec.get("id") for rec in les(root) if rec.get("type") == "beslutning"}
if record["korrigerer"] not in kjente:
raise BeslutningError(
"korrigerer=%r naming no decision in the log; a correction "
"restates a decision that exists" % (record["korrigerer"],)
)
_krev_ubrukt_id(root, record["id"])
return jsonl.append_line(sti(root), record)
def sak_id_for(record):
"""The sak-id a `ja` hands off to `sak`, or ``None`` for a `nei`.
Derived through :func:`paths.sak_id`, so the Norwegian folds and the NFC
normalisation are defined in one place (risk H8) rather than twice.
"""
if record.get("beslutning") != "ja":
return None
dato = record.get("dato") or ""
return paths.sak_id(dato[:7], record.get("arbeidsgiver"), record.get("tittel"))
def speil_i_sakslogg(root, sak_id, record):
"""Mirror a `ja` into the case log, where it is the move to `soker`.
build-brief 5.7 already mirrors outcome records into the case log; this is
the same movement at the other end of the case. The record carries the
decision's own id, so the case log points back into the decision log.
"""
if record.get("beslutning") != "ja":
raise BeslutningError(
"only a `ja` is mirrored into a case log; a `nei` creates no case"
)
paths.validate_sak_id(sak_id)
speil = {
"ts": record["dato"],
"type": "beslutning",
"skjema": jsonl.SKJEMA,
"beslutning": "ja",
"id": record["id"],
"notat": record.get("notat"),
}
return jsonl.append_line(
paths.safe_join(root, "saker", sak_id, "logg.jsonl"), speil
)
def _bygg_beslutning(felt):
_krev_kart(felt)
ukjente = sorted(set(felt) - set(FELTREKKEFOLGE))
if ukjente:
raise BeslutningError(
"unknown field(s) %r; build-brief 5.4 names %r"
% (ukjente, [f for f in FELTREKKEFOLGE if f not in ("type", "skjema")])
)
record = {"type": "beslutning", "skjema": jsonl.SKJEMA}
record["id"] = _tekst(felt, "id")
record["dato"] = _tidspunkt(felt, "dato")
record["kilde"] = _tekst(felt, "kilde")
record["url"] = _tekst_eller_null(felt, "url")
record["tittel"] = _tekst(felt, "tittel")
record["arbeidsgiver"] = _tekst(felt, "arbeidsgiver")
record["beslutning"] = _valg(felt, "beslutning", DECISIONS)
record["arsak"] = _arsak(felt.get("arsak"))
record["notat"] = _notat(felt.get("notat"))
record["score_da"] = _heltall(felt, "score_da")
record["delscore"] = _kriterietabell(felt, "delscore")
record["vekter"] = _kriterietabell(felt, "vekter")
_samme_kriterier(record["delscore"], record["vekter"])
record["vekt_hash"] = _vekt_hash(felt.get("vekt_hash"))
korrigerer = felt.get("korrigerer")
if korrigerer is not None:
raise BeslutningError(
"korrigerer=%r on a decision; a decision that restates another is "
"a `korrigering` record, written with korriger()" % (korrigerer,)
)
record["korrigerer"] = None
return record
def _bygg_korrigering(felt):
_krev_kart(felt)
tillatt = set(KORRIGERING_HODE) | set(KORRIGERBARE)
ukjente = sorted(set(felt) - tillatt)
if ukjente:
raise BeslutningError(
"a correction cannot restate %r; it may restate %r"
% (ukjente, list(KORRIGERBARE))
)
record = {"type": "korrigering", "skjema": jsonl.SKJEMA}
record["id"] = _tekst(felt, "id")
record["dato"] = _tidspunkt(felt, "dato")
korrigerer = felt.get("korrigerer")
if not isinstance(korrigerer, str) or not korrigerer.strip():
raise BeslutningError("korrigerer must name the decision being restated")
record["korrigerer"] = korrigerer
restatert = [navn for navn in KORRIGERBARE if navn in felt]
if not restatert:
raise BeslutningError(
"a correction that restates nothing corrects nothing; name at "
"least one of %r" % (list(KORRIGERBARE),)
)
for navn in restatert:
if navn == "arsak":
record["arsak"] = _arsak(felt.get("arsak"))
elif navn == "notat":
record["notat"] = _notat(felt.get("notat"))
elif navn == "beslutning":
record["beslutning"] = _valg(felt, "beslutning", DECISIONS)
elif navn == "url":
record["url"] = _tekst_eller_null(felt, "url")
else:
record[navn] = _tekst(felt, navn)
return record
def _krev_ubrukt_id(root, ident):
brukte = {rec.get("id") for rec in les(root)}
if ident in brukte:
raise BeslutningError(
"id %r is already in the log. Ids are how a correction names its "
"target, so a reused one makes that reference ambiguous." % (ident,)
)
def _krev_kart(felt):
if not isinstance(felt, dict):
raise BeslutningError(
"a decision must be an object, got %s" % type(felt).__name__
)
def _tekst(felt, navn):
verdi = felt.get(navn)
if not isinstance(verdi, str) or not verdi.strip():
raise BeslutningError("%s must be a non-empty string, got %r" % (navn, verdi))
return verdi
def _tekst_eller_null(felt, navn):
verdi = felt.get(navn)
if verdi is None:
return None
if not isinstance(verdi, str) or not verdi.strip():
raise BeslutningError("%s must be a string or null, got %r" % (navn, verdi))
return verdi
def _tidspunkt(felt, navn):
verdi = felt.get(navn)
if not isinstance(verdi, str):
raise BeslutningError("%s must be an ISO-8601 timestamp, got %r" % (navn, verdi))
# Delegated, so the UTC-offset rule is stated once. Norwegian local time
# repeats an hour every October and a naive stamp cannot be ordered after.
try:
jsonl.instant({navn: verdi})
except jsonl.JsonlError as feil:
raise BeslutningError("%s: %s" % (navn, feil))
return verdi
def _valg(felt, navn, lovlige):
verdi = felt.get(navn)
if verdi not in lovlige:
raise BeslutningError("%s must be one of %r, got %r" % (navn, list(lovlige), verdi))
return verdi
def _arsak(verdi):
if not isinstance(verdi, list):
raise BeslutningError(
"arsak must be a list of %d-%d values from the closed enum, got %r"
% (MIN_ARSAKER, MAKS_ARSAKER, verdi)
)
if not MIN_ARSAKER <= len(verdi) <= MAKS_ARSAKER:
raise BeslutningError(
"arsak holds %d value(s); build-brief 5.4 allows %d to %d"
% (len(verdi), MIN_ARSAKER, MAKS_ARSAKER)
)
if len(set(verdi)) != len(verdi):
raise BeslutningError("arsak repeats a value: %r" % (verdi,))
for enkelt in verdi:
if enkelt not in ARSAKER:
raise BeslutningError(
"arsak %r is outside the closed enum. The twelve are: %s"
% (enkelt, ", ".join(ARSAKER))
)
return list(verdi)
def _notat(verdi):
if verdi is None:
return None
if not isinstance(verdi, str):
raise BeslutningError("notat must be text or null, got %r" % (verdi,))
antall = len(verdi.split())
if antall > MAKS_NOTAT_ORD:
raise BeslutningError(
"notat is %d words; build-brief 5.4 allows at most %d"
% (antall, MAKS_NOTAT_ORD)
)
return verdi
def _heltall(felt, navn):
verdi = felt.get(navn)
# bool is an int in Python, and `score_da: True` would be recorded as 1 and
# averaged with real scores by M6.
if isinstance(verdi, bool) or not isinstance(verdi, int):
raise BeslutningError(
"%s must be an integer -- M6 reads it as a value on a scale, and a "
"string that looks like a number sorts as text; got %r" % (navn, verdi)
)
return verdi
def _kriterietabell(felt, navn):
verdi = felt.get(navn)
if not isinstance(verdi, dict) or not verdi:
raise BeslutningError(
"%s must be a non-empty mapping of criterion to integer, got %r"
% (navn, verdi)
)
for kriterium, tall in verdi.items():
if not isinstance(kriterium, str) or not kriterium:
raise BeslutningError("%s has a criterion that is not a name: %r"
% (navn, kriterium))
if isinstance(tall, bool) or not isinstance(tall, int):
raise BeslutningError(
"%s[%r] must be an integer, got %r" % (navn, kriterium, tall)
)
return dict(verdi)
def _samme_kriterier(delscore, vekter):
if set(delscore) != set(vekter):
raise BeslutningError(
"delscore covers %r and vekter covers %r; a sub-score without a "
"weight cannot have reached the score that was recorded"
% (sorted(delscore), sorted(vekter))
)
def _vekt_hash(verdi):
if not isinstance(verdi, str) or not verdi.startswith("sha256:"):
raise BeslutningError(
"vekt_hash must be the `sha256:` fingerprint scoring handed over, "
"got %r. It is not recomputed here; see the module docstring."
% (verdi,)
)
return verdi
# ---------------------------------------------------------------------------
# Command line (the degradation branch's other half)
#
# `beslutning` degrades to telling the operator to run this from a terminal
# when `jobbsok-tools` is absent, and an instruction to run a command line that
# does not exist would make that branch false.
# ---------------------------------------------------------------------------
EXIT_OK = 0
EXIT_FEIL = 2
def build_parser():
import argparse
parser = argparse.ArgumentParser(
prog="beslutninger.py",
description=(
"Append-only beslutningslogg (build-brief 5.4). Legger til en "
"beslutning eller en korrigering, eller skriver ut stroemmen."
),
)
parser.add_argument("--workspace", required=True,
help="arbeidsomraadets rot (ingen underforstaatt standard)")
parser.add_argument("--json", default=None, metavar="OBJEKT",
help="posten som skal legges til, som JSON")
parser.add_argument("--korriger", action="store_true",
help="posten er en korrigering av en tidligere beslutning")
parser.add_argument("--liste", action="store_true",
help="skriv ut stroemmen med korrigeringer anvendt")
return parser
def _som_json(data):
import json as _json
return _json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
def main(argv=None, stdout=None, stderr=None):
import json as _json
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)
if bool(args.json) == bool(args.liste):
stderr.write("beslutninger: velg enten --json <objekt> eller --liste\n")
return EXIT_FEIL
try:
root = paths.workspace_root(args.workspace)
if args.liste:
stdout.write(_som_json(stromme(root)))
else:
felt = _json.loads(args.json)
record = korriger(root, felt) if args.korriger else legg_til(root, felt)
stdout.write(_som_json(record))
except (BeslutningError, paths.WorkspaceError, jsonl.JsonlError, ValueError,
OSError) as feil:
stderr.write("beslutninger: %s\n" % feil)
return EXIT_FEIL
stdout.flush()
return EXIT_OK
if __name__ == "__main__":
import sys as _sys
_sys.exit(main())