refactor: shared agent workflows (F01) and one application registry (F02)

Checkpoint for the 2026-09-11 project review remediation. Work by a previous
agent that ran out of context; committed unchanged as a rollback point before
the remaining findings land on top of it.

F01 — the two agent directories had drifted into different processes
- Six shared workflows under `resume_builder/reference/workflows/`; both
  `.agents/skills/` and `.claude/skills/` are now thin entrypoints that route
  to the same file instead of carrying full copies.
- The `.claude` copies were the stale side: fixed bullet line variants and
  character budgets that the newer process had already dropped. Removing them
  is the point of the change, not collateral.
- Parity/link regression test added; corrected the top-level file map, which
  pointed at a directory that does not exist.

F02 — application status was stored in four places and disagreed
- New registry `job_scout/state/applications.json` (26 records) with
  `application_records.py` (schema, validation, transactional write) and
  `application_tracker.py` (add/update, no quota).
- AGENTS.md and CLAUDE.md no longer keep their own status tables; both point
  at `current_applications.md`, generated from the registry.
- Interview-reached is now independent of outcome, so a withdrawal after an
  interview can record both facts. The old tables are preserved verbatim in
  `job_scout/state/history/application_status_before_2026-09-11.json`.

Verification: `python -m unittest discover -s tests -v` — 18/18 pass.
Spot-checked that the detailed SBB and Kdo Cy notes survived the migration
into the registry rather than being flattened.

Known unresolved, deliberately not invented (see FIX_PROGRESS.md):
Aker BP's application date and its Adjacent-despite-79 classification.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Qej5SBvhcbGXHwLqNWiskM
This commit is contained in:
dennisthiessenandClaude Opus 5 committed 2026-09-11 19:23:13 +02:00
1 parent 0d999ab21b
commit afdef7eb94
33 files changed
+1558 -1768

No files matched your search

@@ -0,0 +1,140 @@
"""Authoritative application state and generated, read-only projections.
Run with --check to detect stale views, or --render after editing the registry.
Outcome events describe what happened; current outcome never erases an interview.
"""
from __future__ import annotations
import argparse
from contextlib import contextmanager
from datetime import date
import json
import math
import os
import tempfile
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
REGISTRY = ROOT / "job_scout/state/applications.json"
VIEW = ROOT / "resume_builder/reference/current_applications.md"
COHORT = ROOT / "job_scout/state/application_cohort.json"
OUTCOMES = {"open", "applied", "paused", "not_proceeding", "role_closed", "closed_no_response", "withdrawn_by_candidate", "rejected_no_interview", "rejected_after_interview", "interviewing", "offer", "accepted"}
def valid_date(value: str) -> str:
if not isinstance(value, str) or date.fromisoformat(value).isoformat() != value:
raise ValueError("expected ISO date YYYY-MM-DD")
return value
def validate(data: dict) -> None:
ids = set()
for record in data["applications"]:
identifier = record.get("id")
if not identifier or identifier in ids:
raise ValueError("application IDs must be nonempty and unique")
ids.add(identifier)
if record.get("outcome") not in OUTCOMES:
raise ValueError(f"{identifier}: invalid outcome")
for key in ("application_date", "rejection_date", "withdrawn"):
if record.get(key):
valid_date(record[key])
for event in record.get("events", []):
if event.get("date"):
valid_date(event["date"])
score = record.get("evidence_fit")
if score is not None and (isinstance(score, bool) or not isinstance(score, (float, int)) or not math.isfinite(score) or not 0 <= score <= 100):
raise ValueError(f"{identifier}: evidence_fit must be finite and between 0 and 100")
if record.get("hard_gate") == "fail" and record.get("fit_class") not in {"stretch", "no-go"}:
raise ValueError(f"{identifier}: failed gate requires Stretch or No-go")
if score is not None and record.get("hard_gate") == "pass" and record.get("fit_class") in {"core", "adjacent"}:
expected = "core" if score >= 75 else "adjacent" if score >= 60 else "stretch"
if record["fit_class"] != expected and not record.get("fit_override_reason"):
raise ValueError(f"{identifier}: classification differs from score; record fit_override_reason")
@contextmanager
def transaction():
"""Serialize local writers; fail clearly rather than overwrite another update."""
lock = REGISTRY.with_suffix(".lock")
try:
descriptor = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
except FileExistsError as exc:
raise ValueError(f"Application registry is locked: {lock}. Check for an active writer before removing a stale lock.") from exc
try:
with os.fdopen(descriptor, "w") as stream:
stream.write(str(os.getpid()))
data = load()
yield data
validate(data)
atomic_write(REGISTRY, json.dumps(data, indent=2, ensure_ascii=False) + "\n")
render(data)
finally:
lock.unlink(missing_ok=True)
def atomic_write(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
descriptor, temporary = tempfile.mkstemp(dir=path.parent, prefix=path.name + ".", suffix=".tmp")
try:
with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream:
stream.write(text)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, path)
finally:
Path(temporary).unlink(missing_ok=True)
def load() -> dict:
return json.loads(REGISTRY.read_text(encoding="utf-8"))
def interview_reached(record: dict) -> bool:
return bool(record.get("interview")) or any(
event.get("type") in {"interview_invited", "interview_held"}
for event in record.get("events", [])
)
def projections(data: dict) -> dict[Path, str]:
lines = ["# Current applications", "", "Generated from `job_scout/state/applications.json`; do not edit this view.",
"Use current outcome and next action below. Imported legacy notes are historical, not current instructions.", "",
"| Application | Current outcome | Interview reached | Next action |", "|---|---|---|---|"]
for record in data["applications"]:
cells = [record["label"], record["outcome"], "yes" if interview_reached(record) else "not recorded", record["next_action"]]
lines.append("| " + " | ".join(str(cell).replace("|", " / ").replace("\n", " ") for cell in cells) + " |")
lines.extend(["", "## Current targeting notes", "", *["- " + note for note in data.get("targeting_notes", [])], ""])
cohort = dict(data["cohort_metadata"])
cohort["generated_from"] = "job_scout/state/applications.json"
cohort["applications"] = [record for record in data["applications"] if record.get("cohort_id") == cohort["cohort_id"]]
return {VIEW: "\n".join(lines), COHORT: json.dumps(cohort, indent=2, ensure_ascii=False) + "\n"}
def render(data: dict, check: bool = False) -> list[str]:
validate(data)
stale = []
for path, expected in projections(data).items():
if not path.exists() or path.read_text(encoding="utf-8") != expected:
stale.append(str(path.relative_to(ROOT)))
if not check:
atomic_write(path, expected)
return stale
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
mode = parser.add_mutually_exclusive_group(required=True)
mode.add_argument("--check", action="store_true")
mode.add_argument("--render", action="store_true")
args = parser.parse_args()
stale = render(load(), check=args.check)
if args.check and stale:
print("STALE: " + ", ".join(stale))
return 1
print("Application views are current.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,117 @@
"""Record applications without quotas; preserve outcome events and interview history."""
from __future__ import annotations
import argparse
import hashlib
import application_records as records
FIT_CLASSES = ("core", "adjacent", "stretch", "no-go")
CHANNELS = ("strong", "moderate", "weak")
def summary(data: dict, all_applications: bool = False) -> None:
cohort_id = data["cohort_metadata"]["cohort_id"]
applications = [row for row in data["applications"] if all_applications or row.get("cohort_id") == cohort_id]
print(f"Application record: {'all' if all_applications else cohort_id} ({len(applications)} applications; no quota)")
for field, values in (("fit_class", FIT_CLASSES), ("channel", CHANNELS)):
for value in values:
print(f" {field} {value}: {sum(row.get(field) == value for row in applications)}")
unknown = sum(row.get(field) is None for row in applications)
if unknown:
print(f" {field} unrecorded: {unknown}")
outcomes = sorted({row["outcome"] for row in applications})
print(" outcomes: " + ", ".join(f"{value}={sum(row['outcome'] == value for row in applications)}" for value in outcomes))
print(f" interview reached (including later withdrawal/rejection): {sum(records.interview_reached(row) for row in applications)}")
def add_application(data: dict, args: argparse.Namespace) -> None:
records.valid_date(args.date)
identity = args.id or "app-" + hashlib.sha256(f"{args.company.casefold()}|{args.job_id or args.role.casefold()}|{args.date}".encode()).hexdigest()[:12]
candidate = {
"id": identity, "company": args.company, "role": args.role,
"label": f"{args.company} — {args.role}", "application_date": args.date,
"fit_class": args.fit_class, "evidence_fit": args.evidence_fit,
"hard_gate": args.hard_gate, "channel": args.channel, "outcome": args.outcome,
"cohort_id": data["cohort_metadata"]["cohort_id"], "job_id": args.job_id,
"url": args.url, "fit_override_reason": args.fit_override_reason,
"rubric_version": args.rubric_version,
"next_action": args.next_action, "current_summary": args.outcome,
"events": [{"type": args.outcome, "date": args.date}],
}
for existing in data["applications"]:
same_job = args.job_id and existing.get("job_id") == args.job_id and existing["company"].casefold() == args.company.casefold()
same_url = args.url and existing.get("url") == args.url
same_application = existing["company"].casefold() == args.company.casefold() and existing["role"].casefold() == args.role.casefold() and existing.get("application_date") == args.date
if existing["id"] == identity or same_job or same_url or same_application:
if all(existing.get(key) == value for key, value in candidate.items() if key not in {"events", "id", "current_summary", "next_action", "outcome"}):
return
raise ValueError(f"Application already exists ({existing['id']}); use update instead")
data["applications"].append(candidate)
records.validate(data)
def update_application(data: dict, args: argparse.Namespace) -> None:
records.valid_date(args.date)
matches = [row for row in data["applications"] if row["id"] == args.id]
if not matches:
raise ValueError(f"Unknown application ID: {args.id}")
row = matches[0]
if row.get("application_date") and args.date < row["application_date"]:
raise ValueError("event date precedes application date")
outcome_dates = [event["date"] for event in row.get("events", []) if event.get("outcome") and event.get("date")]
if args.outcome and outcome_dates and args.date < max(outcome_dates):
raise ValueError("older outcome cannot replace current state; record it as a historical note")
event = {"type": args.event or args.outcome or "note", "date": args.date}
if args.note:
event["note"] = args.note
if args.outcome:
event["outcome"] = args.outcome
if event not in row.setdefault("events", []):
row["events"].append(event)
if args.outcome:
row["outcome"] = args.outcome
row["current_summary"] = args.note or args.outcome
row["next_action"] = args.next_action or ("Await response." if args.outcome in {"applied", "interviewing", "open"} else "Review recorded outcome before further action.")
elif args.next_action:
row["next_action"] = args.next_action
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
commands = parser.add_subparsers(dest="command", required=True)
view = commands.add_parser("summary")
view.add_argument("--all", action="store_true")
add = commands.add_parser("add")
for field in ("company", "role", "date"):
add.add_argument("--" + field, required=True)
add.add_argument("--id")
add.add_argument("--job-id")
add.add_argument("--url")
add.add_argument("--fit-class", choices=FIT_CLASSES, required=True)
add.add_argument("--evidence-fit", type=float, required=True)
add.add_argument("--channel", choices=CHANNELS, required=True)
add.add_argument("--hard-gate", choices=("pass", "fail", "unknown"), required=True)
add.add_argument("--outcome", choices=sorted(records.OUTCOMES), default="applied")
add.add_argument("--fit-override-reason")
add.add_argument("--rubric-version", default="critique-framework-2026-09-11")
add.add_argument("--next-action", default="Await response.")
update = commands.add_parser("update")
update.add_argument("--id", required=True)
update.add_argument("--date", required=True, help="Date of this event")
update.add_argument("--outcome", choices=sorted(records.OUTCOMES))
update.add_argument("--event", choices=("interview_invited", "interview_held", "assessment_passed", "contact_attempted", "contact_conversation", "referral_received", "note"))
update.add_argument("--note")
update.add_argument("--next-action")
args = parser.parse_args()
try:
if args.command == "summary":
summary(records.load(), args.all)
return
with records.transaction() as data:
(add_application if args.command == "add" else update_application)(data, args)
summary(data)
except (ValueError, OSError) as error:
parser.exit(2, f"ERROR: {error}\n")
if __name__ == "__main__":
main()
+3 -79
View File
@@ -1,81 +1,5 @@
#!/usr/bin/env python3
"""Record and summarize the controlled ten-application cohort."""
"""Compatibility entry point; application_tracker owns the uncapped registry workflow."""
from application_tracker import main
from __future__ import annotations
import argparse
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
TRACKER = ROOT / "job_scout" / "state" / "application_cohort.json"
FIT_CLASSES = ("core", "adjacent", "stretch")
CHANNELS = ("strong", "moderate", "weak")
def load() -> dict:
return json.loads(TRACKER.read_text(encoding="utf-8"))
def save(data: dict) -> None:
TRACKER.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
def summary(data: dict) -> None:
applications = data["applications"]
# Slot caps were dropped 2026-09-05 (application_strategy.md). This is a record, not a
# budget: report counts, never counts-against-a-target, so nothing reads as a remaining slot.
print(f"Application record: {data['cohort_id']} ({len(applications)} applications)")
for fit_class in FIT_CLASSES:
count = sum(item["fit_class"] == fit_class for item in applications)
print(f" {fit_class}: {count}")
for channel in CHANNELS:
count = sum(item["channel"] == channel for item in applications)
print(f" channel {channel}: {count}")
outcomes: dict[str, int] = {}
for item in applications:
outcome = item.get("outcome", "open")
outcomes[outcome] = outcomes.get(outcome, 0) + 1
print(" outcomes: " + ", ".join(f"{key}={value}" for key, value in sorted(outcomes.items())))
def main() -> None:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("summary")
add = subparsers.add_parser("add")
add.add_argument("--company", required=True)
add.add_argument("--role", required=True)
add.add_argument("--date", required=True)
add.add_argument("--fit-class", choices=FIT_CLASSES, required=True)
add.add_argument("--evidence-fit", type=float, required=True)
add.add_argument("--channel", choices=CHANNELS, required=True)
add.add_argument("--hard-gate", choices=("pass", "fail"), required=True)
add.add_argument("--outcome", default="open")
args = parser.parse_args()
data = load()
if args.command == "add":
if len(data["applications"]) >= 10:
raise SystemExit("cohort already contains ten applications")
if args.hard_gate == "fail" and args.fit_class != "stretch":
raise SystemExit("a hard-gate failure must be recorded as stretch")
data["applications"].append(
{
"company": args.company,
"role": args.role,
"application_date": args.date,
"fit_class": args.fit_class,
"evidence_fit": args.evidence_fit,
"hard_gate": args.hard_gate,
"channel": args.channel,
"outcome": args.outcome,
}
)
save(data)
summary(data)
if __name__ == "__main__":
if __name__ == '__main__':
main()
@@ -105,7 +105,7 @@ def workflow_checks() -> list[str]:
"<= 3 lines white space",
"20-21 variable bullets",
],
ROOT / ".agents" / "skills" / "critique" / "SKILL.md": [
ROOT / "resume_builder" / "reference" / "workflows" / "critique.md": [
"Publications 10%",
"Eight-Dimension Scoring",
],