profit/lakehouse
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Phase 23 — contract terms + staffer identity + competence-weighted retrieval

Browse Source 
Matrix-index the "who handled this" dimension so top staffers become
the training signal and juniors inherit their playbooks automatically
via the boost pipeline. Auto-discovered indicators emerge from
comparing trajectories across staffers on similar contracts — that was
always the architectural point; this wires the last piece.

ContractTerms:
- deadline, budget_total_usd, budget_per_hour_max, local_bonus_per_hour,
  local_bonus_radius_mi, fill_requirement ("paramount" | "preferred")
- Attached to ScenarioSpec, propagated into T3 checkpoint + cloud
  rescue prompts so cloud reasons about trade-offs (pivot within bonus
  radius first; respect per-hour cap; split across cities when
  fill_requirement=paramount).

Staffer:
- {id, name, tenure_months, role: senior|mid|junior|trainee}
- On ScenarioSpec; logged at scenario start; attached to KB outcome
- Recomputed StafferStats written to data/_kb/staffers.jsonl after
  every run: total_runs, fill_rate, avg_turns, avg_citations,
  rescue_rate, competence_score.
- Competence formula: 0.45*fill_rate + 0.20*turn_efficiency +
  0.20*citation_density + 0.15*rescue_rate. Normalized to 0..1.

findNeighbors now returns weighted_score = cosine × best_staffer_competence
(floored at 0.3 so high-similarity low-competence neighbors still
surface). pathway_recommender prompt shows the top staffer's identity
so cloud knows WHOSE playbook it's synthesizing from.

Demo infrastructure:
- tests/multi-agent/gen_staffer_demo.ts: 4 personas (Maria senior,
  James mid, Sam junior, Alex trainee) × 3 contracts (Nashville Welder,
  Joliet Warehouse, Indianapolis Assembly). 12 scenarios total.
- scripts/run_staffer_demo.sh: runs the 12 sequentially with
  LH_OVERVIEW_CLOUD=1. Post-run calls kb_staffer_report.py.
- scripts/kb_staffer_report.py: leaderboard + cross-staffer worker
  overlap (names endorsed by ≥2 staffers → auto-discovered high-value
  workers). Top vs bottom differential.

gen_scenarios.ts (Phase 22 generator) also now emits contract terms
on 70% of generated specs — future KB batches populate with realistic
constraint patterns instead of bare role+city+count.

Stress scenario from item A intentionally NOT the production test.
Real staffing has constraints; Nashville contract + staffer demo is
the honest test of whether the architecture produces measurable
differential between coordinator skill levels.

Demo batch launched — 12 runs × ~3min each ≈ 40min unattended. Report
emitted after batch.
This commit is contained in:
root
2026-04-20 22:16:09 -05:00
parent a7fc8e2256
commit 6b71c8e9b2
41 changed files with 2316 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/multi-agent/kb.ts
+158 -6
View File
@@ -34,6 +34,7 @@ const OUTCOMES_FILE = "outcomes.jsonl";
const CONFIG_SNAPSHOTS_FILE = "config_snapshots.jsonl";
const ERROR_CORRECTIONS_FILE = "error_corrections.jsonl";
const RECOMMENDATIONS_FILE = "pathway_recommendations.jsonl";
const STAFFERS_FILE = "staffers.jsonl";
// What a playbook signature looks like — stable hash of the scenario
// shape that ignores timestamps and specific worker IDs. Two runs with
@@ -60,11 +61,19 @@ export interface RunOutcome {
overview: string;
overview_cloud: boolean;
};
staffer?: { // Phase 23 — who ran this scenario
id: string;
name: string;
tenure_months: number;
role: string;
};
ok_events: number;
total_events: number;
total_turns: number;
total_gap_signals: number;
total_citations: number;
rescue_attempts: number; // Phase 22 item B
rescue_successes: number;
per_event: Array<{
at: string;
kind: string;
@@ -78,6 +87,27 @@ export interface RunOutcome {
created_at: string;
}
// Per-staffer aggregate. Recomputed from outcomes.jsonl after every
// run. Competence_score weights neighbor retrieval so Senior staffers'
// playbooks rank higher for similar scenarios than Junior staffers'.
export interface StafferStats {
id: string;
name: string;
tenure_months: number;
role: string;
total_runs: number;
total_events_attempted: number;
total_events_ok: number;
fill_rate: number; // ok / attempted
avg_turns_per_event: number;
avg_citations_per_run: number;
rescue_attempts: number;
rescue_successes: number;
rescue_rate: number;
competence_score: number; // 0.0 - 1.0, see formula below
last_updated: string;
}
// The AI-synthesized recommendation written back for future runs.
export interface PathwayRecommendation {
sig_hash: string;
@@ -142,7 +172,7 @@ async function readJsonl<T>(file: string): Promise<T[]> {
// embedding, appends outcome, updates or inserts signature entry.
export async function indexRun(
scenarioDir: string,
spec: { client: string; date: string; events: Array<any> },
spec: { client: string; date: string; events: Array<any>; staffer?: { id: string; name: string; tenure_months: number; role: string } },
models: { executor: string; reviewer: string; overview: string; overview_cloud: boolean },
elapsed_secs: number,
): Promise<{ sig_hash: string; outcome: RunOutcome }> {
@@ -154,16 +184,21 @@ export async function indexRun(
// Read results.json produced by scenario.ts to build the outcome record.
const resultsRaw = await readFile(join(scenarioDir, "results.json"), "utf8");
const results = JSON.parse(resultsRaw) as any[];
const rescue_attempts = results.filter(r => r.retry_remediation).length;
const rescue_successes = results.filter(r => r.retry_remediation && r.ok).length;
const outcome: RunOutcome = {
sig_hash,
run_id: scenarioDir.split("/").pop() ?? scenarioDir,
date: spec.date,
models,
staffer: spec.staffer,
ok_events: results.filter(r => r.ok).length,
total_events: results.length,
total_turns: results.reduce((s, r) => s + (r.turns ?? 0), 0),
total_gap_signals: results.reduce((s, r) => s + (r.gap_signals?.length ?? 0), 0),
total_citations: results.reduce((s, r) => s + (r.playbook_citations?.length ?? 0), 0),
rescue_attempts,
rescue_successes,
per_event: results.map(r => ({
at: r.event.at,
kind: r.event.kind,
@@ -178,6 +213,13 @@ export async function indexRun(
};
await appendFile(join(KB_DIR, OUTCOMES_FILE), JSON.stringify(outcome) + "\n");
// Phase 23 — recompute this staffer's aggregate + competence_score.
// Cheap until we have thousands of runs; swap to streaming aggregate
// when that becomes a concern.
if (spec.staffer) {
await recomputeStafferStats(spec.staffer.id);
}
// Embed + upsert signature. Skip if embedding fails — the outcome is
// still persisted, just without neighbor-search hookup.
try {
@@ -221,6 +263,72 @@ export async function indexRun(
return { sig_hash, outcome };
}
// Phase 23 — per-staffer aggregate, recomputed from scratch on each
// run. Writes the full table to STAFFERS_FILE (not append-only; we
// always want the current-state snapshot). Competence score is bounded
// 0..1 and combines four dimensions:
// fill_rate — how often the staffer completes their events
// turn_efficiency — lower turn counts are better
// citation_density — signals use of playbook_memory feedback
// rescue_rate — cloud rescue successes per attempt (doesn't
// penalize staffers who never needed one)
// Weights: fill 0.45, turn_eff 0.20, citation 0.20, rescue 0.15.
// Rationale: completing the job matters most; everything else is style.
export async function recomputeStafferStats(staffer_id: string): Promise<StafferStats | null> {
const outcomes = await readJsonl<RunOutcome>(OUTCOMES_FILE);
const mine = outcomes.filter(o => o.staffer?.id === staffer_id);
if (mine.length === 0) return null;
const latest = mine[mine.length - 1].staffer!;
const total_events_attempted = mine.reduce((s, o) => s + o.total_events, 0);
const total_events_ok = mine.reduce((s, o) => s + o.ok_events, 0);
const total_turns = mine.reduce((s, o) => s + o.total_turns, 0);
const total_citations = mine.reduce((s, o) => s + o.total_citations, 0);
const rescue_attempts = mine.reduce((s, o) => s + (o.rescue_attempts ?? 0), 0);
const rescue_successes = mine.reduce((s, o) => s + (o.rescue_successes ?? 0), 0);
const fill_rate = total_events_attempted > 0 ? total_events_ok / total_events_attempted : 0;
const avg_turns_per_event = total_events_attempted > 0 ? total_turns / total_events_attempted : 0;
const avg_citations_per_run = mine.length > 0 ? total_citations / mine.length : 0;
const rescue_rate = rescue_attempts > 0 ? rescue_successes / rescue_attempts : 1.0; // never-rescued defaults to perfect
// Normalize each axis to 0..1. Caps chosen from observed distributions.
const turn_eff = avg_turns_per_event === 0 ? 0 : Math.max(0, 1 - Math.min(avg_turns_per_event / 10, 1));
const cite_norm = Math.min(avg_citations_per_run / 3, 1);
const competence_score = 0.45 * fill_rate
+ 0.20 * turn_eff
+ 0.20 * cite_norm
+ 0.15 * rescue_rate;
const stats: StafferStats = {
id: staffer_id,
name: latest.name,
tenure_months: latest.tenure_months,
role: latest.role,
total_runs: mine.length,
total_events_attempted,
total_events_ok,
fill_rate,
avg_turns_per_event,
avg_citations_per_run,
rescue_attempts,
rescue_successes,
rescue_rate,
competence_score,
last_updated: new Date().toISOString(),
};
// Rewrite whole file — cheap at O(staffers) scale.
const existing = await readJsonl<StafferStats>(STAFFERS_FILE);
const others = existing.filter(s => s.id !== staffer_id);
others.push(stats);
await writeFile(
join(KB_DIR, STAFFERS_FILE),
others.map(s => JSON.stringify(s)).join("\n") + "\n",
);
return stats;
}
// Public accessor — used by findNeighbors + the staffer report script.
export async function loadStafferStats(): Promise<StafferStats[]> {
return readJsonl<StafferStats>(STAFFERS_FILE);
}
// Cosine similarity for neighbor lookup. Float32-in-memory is fine for
// O(thousands) signatures; swap to vectord HNSW once the corpus grows.
function cosine(a: number[], b: number[]): number {
@@ -237,9 +345,18 @@ function cosine(a: number[], b: number[]): number {
// Find the k nearest-neighbor signatures for a target spec, returning
// neighbor sigs + their outcome history. This is what the recommender
// feeds to the overview model.
//
// Phase 23 — weighted by staffer competence. The ranking score is
// cosine_similarity * max_competence_among_this_sig's_outcomes. Top
// staffers' playbooks surface first even if their scenario was
// slightly less similar than a junior's. Sigs with no staffer history
// fall back to raw similarity (competence defaults to 0.5).
export async function findNeighbors(spec: any, k = 5): Promise<Array<{
sig: PlaybookSignature;
similarity: number;
weighted_score: number;
best_staffer_competence: number;
best_staffer_id: string | null;
outcomes: RunOutcome[];
}>> {
await ensureKb();
@@ -256,16 +373,45 @@ export async function findNeighbors(spec: any, k = 5): Promise<Array<{
const sigs = await readJsonl<PlaybookSignature>(SIGNATURES_FILE);
const outcomes = await readJsonl<RunOutcome>(OUTCOMES_FILE);
const staffers = await readJsonl<StafferStats>(STAFFERS_FILE);
const competenceById = new Map(staffers.map(s => [s.id, s.competence_score]));
// Per-sig best staffer — the highest-competence coordinator who has
// ever run this signature. Used to weight the sig's ranking.
function bestStafferFor(sig_hash: string): { id: string | null; competence: number } {
const mine = outcomes.filter(o => o.sig_hash === sig_hash && o.staffer?.id);
if (mine.length === 0) return { id: null, competence: 0.5 };
let bestId: string | null = null;
let bestComp = 0;
for (const o of mine) {
const c = competenceById.get(o.staffer!.id) ?? 0.5;
if (c > bestComp) { bestComp = c; bestId = o.staffer!.id; }
}
return { id: bestId, competence: bestComp };
}
const ranked = sigs
.filter(s => s.sig_hash !== targetHash) // don't recommend against self
.map(s => ({ sig: s, similarity: cosine(targetVec, s.embedding) }))
.sort((a, b) => b.similarity - a.similarity)
.map(s => {
const similarity = cosine(targetVec, s.embedding);
const best = bestStafferFor(s.sig_hash);
// Weighted score: similarity multiplied by the best staffer's
// competence (floored at 0.3 so a sig with a low-competence
// staffer still shows up if similarity is very high).
const weight = Math.max(best.competence, 0.3);
return {
sig: s,
similarity,
weighted_score: similarity * weight,
best_staffer_competence: best.competence,
best_staffer_id: best.id,
};
})
.sort((a, b) => b.weighted_score - a.weighted_score)
.slice(0, k);
return ranked.map(r => ({
sig: r.sig,
similarity: r.similarity,
...r,
outcomes: outcomes.filter(o => o.sig_hash === r.sig.sig_hash),
}));
}
@@ -342,6 +488,8 @@ export async function recommendFor(
// Build the prompt. Include target spec digest, neighbor digests with
// their outcome stats, and recent error corrections. Ask for
// structured output so we can parse it.
const staffers = await readJsonl<StafferStats>(STAFFERS_FILE);
const stafferById = new Map(staffers.map(s => [s.id, s]));
const neighborBlock = neighbors.map(n => {
const best = n.outcomes.reduce((a, b) =>
a && a.ok_events / Math.max(1, a.total_events) >= b.ok_events / Math.max(1, b.total_events) ? a : b,
@@ -349,7 +497,11 @@ export async function recommendFor(
const avgOk = n.outcomes.length > 0
? (n.outcomes.reduce((s, o) => s + o.ok_events, 0) / n.outcomes.length).toFixed(1)
: "?";
return `- sig ${n.sig.sig_hash} sim=${n.similarity.toFixed(3)} (${n.sig.events_digest.slice(0, 100)}): ${n.outcomes.length} runs, avg ${avgOk}/${best?.total_events ?? "?"} ok; best models: exec=${best?.models.executor ?? "?"} review=${best?.models.reviewer ?? "?"}`;
const topStaffer = n.best_staffer_id ? stafferById.get(n.best_staffer_id) : null;
const stafferTag = topStaffer
? ` [top staffer: ${topStaffer.name} (${topStaffer.role}, competence=${topStaffer.competence_score.toFixed(2)})]`
: "";
return `- sig ${n.sig.sig_hash} sim=${n.similarity.toFixed(3)} weighted=${n.weighted_score.toFixed(3)}${stafferTag}: ${n.outcomes.length} runs, avg ${avgOk}/${best?.total_events ?? "?"} ok; best models: exec=${best?.models.executor ?? "?"} review=${best?.models.reviewer ?? "?"}`;
}).join("\n");
const correctionBlock = corrections.slice(-3).map(c =>