refactor: deliver raw GrowEvents to QScore

This commit is contained in:
Sai-karthik
2026-07-13 20:14:14 +00:00
parent e444d1be15
commit c1d2acd96d
28 changed files with 667 additions and 1191 deletions
+21
View File
@@ -0,0 +1,21 @@
CREATE TABLE IF NOT EXISTS "event_outbox" (
"id" text PRIMARY KEY DEFAULT gen_random_uuid()::text NOT NULL,
"topic" text NOT NULL,
"aggregate_id" text NOT NULL,
"destination" text NOT NULL,
"payload" jsonb NOT NULL,
"headers" jsonb DEFAULT '{}'::jsonb NOT NULL,
"status" text DEFAULT 'pending' NOT NULL,
"attempts" integer DEFAULT 0 NOT NULL,
"available_at" timestamp with time zone DEFAULT now() NOT NULL,
"locked_at" timestamp with time zone,
"published_at" timestamp with time zone,
"last_error" text,
"response" jsonb,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
"updated_at" timestamp with time zone DEFAULT now() NOT NULL,
CONSTRAINT "event_outbox_status_check" CHECK ("status" IN ('pending', 'processing', 'published'))
);
CREATE INDEX IF NOT EXISTS "event_outbox_delivery_idx" ON "event_outbox" USING btree ("status", "available_at");
CREATE UNIQUE INDEX IF NOT EXISTS "event_outbox_topic_aggregate_idx" ON "event_outbox" USING btree ("topic", "aggregate_id");
+7
View File
@@ -99,6 +99,13 @@
"when": 1783801800000,
"tag": "0013_onboarding",
"breakpoints": true
},
{
"idx": 14,
"version": "7",
"when": 1783987200000,
"tag": "0014_event_outbox",
"breakpoints": true
}
]
}
+1
View File
@@ -11,6 +11,7 @@
"test:passive-actions": "tsx scripts/passive-actions.test.ts",
"test:curator-static": "tsx scripts/curator-static-registry.test.ts",
"test:curator-reconcile": "tsx scripts/curator-persisted-reconcile.test.ts",
"test:qscore-raw-events": "tsx scripts/qscore-raw-event-contract.test.ts",
"start": "node dist/index.js",
"typecheck": "tsc -p tsconfig.json --noEmit",
"workflows:smoke": "tsx src/workflows/smoke-test.ts",
+20 -64
View File
@@ -1,68 +1,24 @@
import assert from "node:assert/strict";
import { extractQscoreSignals } from "../src/events/projectors/qscore-projector.js";
import type { GrowEventRow } from "../src/db/schema.js";
import { canonicalGrowEventEnvelope } from "../src/events/record-grow-event.js";
import { normalizeGrowEvent } from "../src/events/normalize.js";
/**
* Test: Completed interview and roleplay events MUST emit at least one signal
* even when session_count/scenario_count is absent. A completion event with no
* count and no rubric data should still produce a single-completion signal.
*/
function event(overrides: Partial<GrowEventRow> & { type: string; source: string; payload: Record<string, unknown> }): GrowEventRow {
return {
id: `event-${overrides.type}`,
userId: "user_test",
orgId: null,
source: overrides.source,
type: overrides.type,
/** Bare completions stay bare. QScore may aggregate them as real events, but
* the backend must never invent session_count/scenario_count or a score. */
for (const [source, type] of [
["interview-service", "interview.session.completed"],
["roleplay-service", "roleplay.scenario.completed"],
] as const) {
const event = canonicalGrowEventEnvelope(normalizeGrowEvent({
id: `event-${type}`,
source,
type,
category: "service",
occurredAt: new Date("2026-07-09T00:00:00.000Z"),
receivedAt: new Date("2026-07-09T00:00:01.000Z"),
mission: overrides.mission ?? null,
subject: overrides.subject ?? null,
correlation: overrides.correlation ?? null,
payload: overrides.payload,
raw: {},
dedupeKey: null,
processingStatus: "pending",
processingError: null,
processedAt: null,
};
occurredAt: "2026-07-14T00:00:00.000Z",
payload: {},
}), "real-user");
assert.deepEqual(event.payload, {});
assert.equal("score" in event.payload as object, false);
assert.equal("session_count" in event.payload as object, false);
assert.equal("scenario_count" in event.payload as object, false);
}
// ── Interview completed with NO count and NO rubric ──────────────────────────
const interviewSignals = extractQscoreSignals(event({
source: "interview-service",
type: "interview.session.completed",
payload: {}, // nothing — just a bare completion event
}));
assert.ok(
interviewSignals.length > 0,
"Interview completed event with no count/rubric must emit at least one signal",
);
assert.ok(
interviewSignals.some((s) => s.signalId === "interview.sessions_completed" && s.score > 0),
"Bare interview completion should emit interview.sessions_completed with positive score",
);
// ── Roleplay completed with NO count and NO rubric ───────────────────────────
const roleplaySignals = extractQscoreSignals(event({
source: "roleplay-service",
type: "roleplay.scenario.completed",
payload: {}, // nothing — just a bare completion event
}));
assert.ok(
roleplaySignals.length > 0,
"Roleplay completed event with no count/rubric must emit at least one signal",
);
assert.ok(
roleplaySignals.some((s) => s.signalId === "roleplay.scenarios_completed" && s.score > 0),
"Bare roleplay completion should emit roleplay.scenarios_completed with positive score",
);
console.log("count-fallback tests passed");
process.exit(0);
console.log("no count fallback tests passed");
-64
View File
@@ -1,29 +1,5 @@
import assert from "node:assert/strict";
import { extractQscoreSignals } from "../src/events/projectors/qscore-projector.js";
import { buildMatchmakingGatewayEvent } from "../src/services/matchmaking-events.js";
import type { GrowEventRow } from "../src/db/schema.js";
function event(overrides: Partial<GrowEventRow> & { type: string; payload: Record<string, unknown> }): GrowEventRow {
return {
id: `event-${overrides.type}`,
userId: "user_test",
orgId: null,
source: "matchmaking-v2",
type: overrides.type,
category: "service",
occurredAt: new Date("2026-07-09T00:00:00.000Z"),
receivedAt: new Date("2026-07-09T00:00:01.000Z"),
mission: overrides.mission ?? null,
subject: overrides.subject ?? null,
correlation: overrides.correlation ?? null,
payload: overrides.payload,
raw: {},
dedupeKey: null,
processingStatus: "pending",
processingError: null,
processedAt: null,
};
}
const saved = buildMatchmakingGatewayEvent({
userId: "user_test",
@@ -62,31 +38,6 @@ assert.equal(saved.payload.curatorTaskId, "task_123");
assert.equal(saved.payload.opportunityId, "job_123");
assert.equal(saved.payload.action, "mark_saved");
const generatedSignals = extractQscoreSignals(event({
type: "matchmaking.matches.generated",
payload: {
action: "run_search",
matchCount: 6,
scanned: 120,
result: { status: "completed" },
},
}));
assert.ok(
generatedSignals.some((signal) => signal.signalId === "matching.match_rate" && signal.score > 0),
"generated matches should produce a match rate signal from match counts",
);
const savedSignals = extractQscoreSignals(event({
type: "matchmaking.match.saved",
payload: {
action: "mark_saved",
opportunityId: "job_123",
curatorTaskId: "task_123",
result: { status: "completed" },
},
}));
assert.equal(savedSignals.length, 0, "saved opportunities no longer emit a legacy signal");
const reviewed = buildMatchmakingGatewayEvent({
userId: "user_test",
action: "record_feedback",
@@ -98,20 +49,5 @@ const reviewed = buildMatchmakingGatewayEvent({
});
assert.equal(reviewed.type, "matchmaking.matches.reviewed");
const appliedSignals = extractQscoreSignals(event({
type: "matchmaking.match.applied",
payload: {
action: "mark_applied",
opportunityId: "job_123",
curatorTaskId: "task_123",
applications_submitted: 1,
result: { status: "completed" },
},
}));
assert.ok(
appliedSignals.some((signal) => signal.signalId === "matching.applications_submitted" && signal.score > 0),
"applied opportunities should produce an applications submitted signal",
);
console.log("matchmaking-events tests passed");
process.exit(0);
+72
View File
@@ -0,0 +1,72 @@
import assert from "node:assert/strict";
import { readFile } from "node:fs/promises";
import { canonicalGrowEventEnvelope } from "../src/events/record-grow-event.js";
import { normalizeGrowEvent } from "../src/events/normalize.js";
import {
projectedSignalsFromQscoreResponse,
QSCORE_GROW_EVENT_DESTINATION,
QSCORE_GROW_EVENT_TOPIC,
} from "../src/events/qscore-event-outbox.js";
const input = {
id: "evt-real-1",
userId: "untrusted-body-user",
orgId: "growqr",
source: "interview-service",
type: "interview.review.completed",
category: "service" as const,
occurredAt: "2026-07-14T10:00:00.000Z",
mission: { instanceId: "mission-1", missionId: "interview-to-offer", stageId: "mock-interview" },
subject: { serviceId: "interview", externalId: "session-1" },
correlation: { sessionId: "session-1", taskId: "task-1" },
payload: {
review: {
overall_score: 78,
rubric_scores: { communication: 82, technical_accuracy: 70 },
},
},
dedupeKey: "interview:session-1:review",
};
const normalized = normalizeGrowEvent(input, { userId: "clerk-user-real" });
const canonical = canonicalGrowEventEnvelope(normalized, "clerk-user-real");
assert.deepEqual(canonical, {
...input,
userId: "clerk-user-real",
raw: input,
});
assert.equal(QSCORE_GROW_EVENT_TOPIC, "qscore.grow_event.raw");
assert.equal(QSCORE_GROW_EVENT_DESTINATION, "/v1/events/ingest");
const projected = projectedSignalsFromQscoreResponse({
projected_signals: [
{ signal_id: "interview.overall_score", score: 45, present: true, evidence: { eventId: input.id } },
{ signal_id: "interview.sessions_completed", evidence: { eventId: input.id } },
],
});
assert.deepEqual(projected, [
{ signalId: "interview.overall_score", score: 45, present: true, raw: { eventId: input.id } },
{ signalId: "interview.sessions_completed", raw: { eventId: input.id } },
]);
const projectorSource = await readFile(new URL("../src/events/projectors/qscore-projector.ts", import.meta.url), "utf8");
for (const forbidden of ["clampScore", "BASELINE_SCORE", "extractResumeSignals", "forwardSignalsToQscoreService"]) {
assert.equal(projectorSource.includes(forbidden), false, `projector must not contain ${forbidden}`);
}
const serviceAgentSource = await readFile(new URL("../src/services/service-agents.ts", import.meta.url), "utf8");
assert.equal(/q_score\s*:\s*\d/.test(serviceAgentSource), false, "service agents must not inject numeric QScore data");
const homeSource = await readFile(new URL("../src/home/home-feed.ts", import.meta.url), "utf8");
for (const forbidden of ["|| 47", "score - 29", "Math.max(35", "ensureOnboardingBaselineQscore"]) {
assert.equal(homeSource.includes(forbidden), false, `Home must not contain synthetic QScore fallback ${forbidden}`);
}
const servicesSource = await readFile(new URL("../src/routes/services.ts", import.meta.url), "utf8");
assert.equal(
/app\.get\("\/qscore\/current"[\s\S]*?qscore\.review\.opened/.test(servicesSource),
false,
"polling the current QScore must remain a read and must not emit a synthetic review event",
);
console.log("qscore raw-event ownership contract: ok");
+24 -100
View File
@@ -1,106 +1,30 @@
import assert from "node:assert/strict";
import { extractQscoreSignals } from "../src/events/projectors/qscore-projector.js";
import type { GrowEventRow } from "../src/db/schema.js";
import { canonicalGrowEventEnvelope } from "../src/events/record-grow-event.js";
import { normalizeGrowEvent } from "../src/events/normalize.js";
/**
* Test: Service-ingest projector behavior across multiple service sources.
* Verifies that canonical Grow events from different services produce
* registry-valid signals with correct source attribution.
*/
/** Every service family is forwarded as a canonical event with unchanged real
* payload. Extraction and signal validation now run only in QScore. */
const cases = [
["resume-builder", "resume.analysis.completed", { analysis: { score_breakdown: [{ category: "ATS Compatibility", score: 75 }] } }],
["qscore-service", "qscore.signal.projected", { score: 65 }],
["social-branding-service", "brand.profile.updated", { profile: { headline: "Real headline" } }],
["unknown-service", "service.completed", {}],
["matchmaking-v2", "matchmaking.match.applied", { applications_submitted: 3 }],
] as const;
function event(overrides: Partial<GrowEventRow> & { type: string; source: string; payload: Record<string, unknown> }): GrowEventRow {
return {
id: `event-${overrides.type}`,
userId: "user_test",
orgId: null,
source: overrides.source,
type: overrides.type,
for (const [source, type, payload] of cases) {
const canonical = canonicalGrowEventEnvelope(normalizeGrowEvent({
id: `event-${type}`,
userId: "real-user",
source,
type,
category: "service",
occurredAt: new Date("2026-07-09T00:00:00.000Z"),
receivedAt: new Date("2026-07-09T00:00:01.000Z"),
mission: overrides.mission ?? null,
subject: overrides.subject ?? null,
correlation: overrides.correlation ?? null,
payload: overrides.payload,
raw: {},
dedupeKey: null,
processingStatus: "pending",
processingError: null,
processedAt: null,
};
occurredAt: "2026-07-14T00:00:00.000Z",
payload,
}), "real-user");
assert.equal(canonical.source, source);
assert.equal(canonical.type, type);
assert.deepEqual(canonical.payload, payload);
}
// ── 1. Resume analysis produces resume signals ──────────────────────────────
{
const signals = extractQscoreSignals(event({
source: "resume-builder",
type: "resume.analysis.completed",
payload: { analysis: { score_breakdown: [{ category: "ATS Compatibility", score: 75 }] } },
}));
assert.ok(signals.length > 0, "resume analysis event should produce signals");
assert.ok(
signals.some((s) => s.signalId.startsWith("resume.")),
"resume source should produce resume.* signals",
);
}
// ── 2. Generic scored service event (qscore source) produces completion score ─
{
const signals = extractQscoreSignals(event({
source: "qscore-service",
type: "qscore.signal.projected",
payload: { score: 65 },
}));
assert.ok(
signals.some((s) => s.score === 65),
"qscore source event with score should forward the score",
);
}
// ── 3. Social branding pre-computed signals forwarded as-is ─────────────────
{
const signals = extractQscoreSignals(event({
source: "social-branding-service",
type: "brand.profile.updated",
payload: {
qscore_signals: [
{ signalId: "linkedin.headline_quality", score: 70 },
{ signalId: "linkedin.summary_complete", score: 65 },
],
},
}));
assert.ok(
signals.some((s) => s.signalId === "linkedin.headline_quality" && s.score === 70),
"social branding should forward pre-computed signals as-is",
);
assert.ok(
signals.some((s) => s.signalId === "linkedin.summary_complete"),
"social branding should forward all pre-computed signals",
);
}
// ── 4. Unknown source with no score produces no signals ──────────────────────
{
const signals = extractQscoreSignals(event({
source: "unknown-service",
type: "service.completed",
payload: {},
}));
assert.equal(signals.length, 0, "unknown source with no score should produce no signals");
}
// ── 5. Matchmaking applied event with count ──────────────────────────────────
{
const signals = extractQscoreSignals(event({
source: "matchmaking-v2",
type: "matchmaking.match.applied",
payload: { applications_submitted: 3 },
}));
assert.ok(
signals.some((s) => s.signalId === "matching.applications_submitted" && s.score > 0),
"matchmaking applied with count should produce applications_submitted signal",
);
}
console.log("service-ingest-projector tests passed");
process.exit(0);
console.log("service raw-ingest forwarding tests passed");
+19 -227
View File
@@ -1,238 +1,30 @@
import assert from "node:assert/strict";
import { extractQscoreSignals } from "../src/events/projectors/qscore-projector.js";
import { ONBOARDING_BASELINE_SIGNAL_ID } from "../src/events/onboarding-qscore.js";
import type { GrowEventRow } from "../src/db/schema.js";
import { readFile } from "node:fs/promises";
import { canonicalGrowEventEnvelope } from "../src/events/record-grow-event.js";
import { normalizeGrowEvent } from "../src/events/normalize.js";
/**
* Test: All signal IDs emitted by the projector must be valid members of the
* qscore_service v2 signal registry. The onboarding baseline signal must be
* "onboarding.completed" (not the non-registry "onboarding.completed_baseline").
* Signal registration and all 67 extraction policies belong to QScore. This
* backend contract test prevents the old local registry/projector from being
* reintroduced while proving the complete raw evidence reaches QScore.
*/
// v2 registry — union of all signal_ids across all 7 profession formula JSONs.
// Sourced from qscore_service/app/scoring/formula/v2/*/formula.json
const V2_REGISTRY = new Set<string>([
// resume
"resume.uploaded",
"resume.ats_compatibility",
"resume.keyword_relevance",
"resume.quantified_achievements",
"resume.grammar_clarity",
"resume.format_structure",
"resume.contact_info",
"resume.page_count",
// interview
"interview.sessions_completed",
"interview.overall_score",
"interview.response_clarity",
"interview.technical_accuracy",
"interview.behavioral_quality",
"interview.improvement_over_time",
"interview.type_diversity",
// roleplay
"roleplay.scenarios_completed",
"roleplay.situational_judgment",
"roleplay.empathy_demonstrated",
"roleplay.problem_resolution",
"roleplay.communication_effectiveness",
// courses
"courses.started",
"courses.completed",
"courses.completion_rate",
"courses.difficulty",
"courses.pathway_relevance",
// matching
"matching.jobs_viewed",
"matching.applications_submitted",
"matching.application_quality",
"matching.match_rate",
// onboarding
"onboarding.completed",
"onboarding.initial_skills_score",
"onboarding.goals_clarity",
"onboarding.profile_completeness",
"onboarding.self_assessment_accuracy",
"onboarding.motivation_quality",
// coverletter
"coverletter.uploaded",
"coverletter.customization",
"coverletter.persuasiveness",
// linkedin
"linkedin.account_connected",
"linkedin.profile_photo",
"linkedin.headline_quality",
"linkedin.summary_complete",
"linkedin.experience_detail",
"linkedin.skills_listed",
// portfolio
"portfolio.website_exists",
"portfolio.website_quality",
"portfolio.content_relevance",
// content
"content.articles_count",
"content.quality",
"content.platform_credibility",
// ugc
"ugc.speaking_count",
"ugc.event_credibility",
"ugc.verifiable_evidence",
// assessments
"assessments.taken",
"assessments.passed",
"assessments.avg_score",
"assessments.skill_diversity",
// presentations
"presentations.submitted",
"presentations.content_quality",
"presentations.delivery_score",
"presentations.visual_aids",
// events
"events.webinars_attended",
"events.meetups_attended",
"events.participation_quality",
// mentor
"mentor.feedback_count",
"mentor.score",
"mentor.implementation_rate",
]);
function event(overrides: Partial<GrowEventRow> & { type: string; payload: Record<string, unknown>; source: string }): GrowEventRow {
return {
id: `event-${overrides.type}`,
userId: "user_test",
orgId: null,
source: overrides.source,
type: overrides.type,
category: "service",
occurredAt: new Date("2026-07-09T00:00:00.000Z"),
receivedAt: new Date("2026-07-09T00:00:01.000Z"),
mission: overrides.mission ?? null,
subject: overrides.subject ?? null,
correlation: overrides.correlation ?? null,
payload: overrides.payload,
raw: {},
dedupeKey: null,
processingStatus: "pending",
processingError: null,
processedAt: null,
};
}
// ── 1. Onboarding signal ID must be the registry-valid "onboarding.completed" ─
assert.equal(
ONBOARDING_BASELINE_SIGNAL_ID,
"onboarding.completed",
"Onboarding baseline signal ID must be 'onboarding.completed' (registry-valid), not 'onboarding.completed_baseline'",
);
assert.ok(
V2_REGISTRY.has(ONBOARDING_BASELINE_SIGNAL_ID),
"Onboarding signal ID must appear in v2 registry",
);
// ── 2. All projector signal IDs must be in the registry ──────────────────────
// Construct representative events for each source family and check every emitted
// signal ID is registry-valid.
// Resume with full breakdown
const resumeSignals = extractQscoreSignals(event({
const raw = {
id: "evt-full-evidence",
source: "resume-builder",
type: "resume.analysis.completed",
category: "service" as const,
occurredAt: "2026-07-14T00:00:00.000Z",
payload: {
analysis: {
score_breakdown: [
{ category: "ATS Compatibility", score: 70 },
{ category: "Content Quality", score: 65 },
{ category: "Formatting", score: 80 },
],
dimensional_scores: [
{ dimension: "Keywords", score: 60 },
{ dimension: "Quantification", score: 55 },
],
score_breakdown: [{ category: "ATS Compatibility", score: 70 }],
dimensional_scores: [{ dimension: "Keywords", score: 60 }],
},
},
}));
for (const s of resumeSignals) {
assert.ok(
V2_REGISTRY.has(s.signalId),
`Resume signal '${s.signalId}' is not in the v2 registry`,
);
}
};
const event = canonicalGrowEventEnvelope(normalizeGrowEvent(raw), "real-user");
assert.deepEqual(event.payload, raw.payload, "raw evidence must reach QScore without signal extraction");
// Interview with rubric
const interviewSignals = extractQscoreSignals(event({
source: "interview-service",
type: "interview.session.completed",
payload: {
review: {
overall_score: 72,
rubric_scores: { content_quality: 70, role_alignment: 68, language: 65 },
historical_comparison: { overall_delta: 5 },
},
session_count: 3,
type_diversity: 2,
},
}));
for (const s of interviewSignals) {
assert.ok(
V2_REGISTRY.has(s.signalId),
`Interview signal '${s.signalId}' is not in the v2 registry`,
);
}
// Roleplay with rubric
const roleplaySignals = extractQscoreSignals(event({
source: "roleplay-service",
type: "roleplay.scenario.completed",
payload: {
review: {
rubric_scores: {
scenario_adherence: 75,
emotional_intelligence: 70,
adaptability: 68,
content: 72,
},
},
scenario_count: 4,
},
}));
for (const s of roleplaySignals) {
assert.ok(
V2_REGISTRY.has(s.signalId),
`Roleplay signal '${s.signalId}' is not in the v2 registry`,
);
}
// Courses
const courseSignals = extractQscoreSignals(event({
source: "courses-service",
type: "course.completed",
payload: {
courseId: "c1",
completed_count: 2,
watchPct: 0.9,
difficulty: "intermediate",
label: "high relevance",
},
}));
for (const s of courseSignals) {
assert.ok(
V2_REGISTRY.has(s.signalId),
`Course signal '${s.signalId}' is not in the v2 registry`,
);
}
// Matchmaking
const matchSignals = extractQscoreSignals(event({
source: "matchmaking-v2",
type: "matchmaking.feed.viewed",
payload: { jobs_viewed: 25 },
}));
for (const s of matchSignals) {
assert.ok(
V2_REGISTRY.has(s.signalId),
`Matchmaking signal '${s.signalId}' is not in the v2 registry`,
);
}
console.log("signal-registry tests passed");
process.exit(0);
const projector = await readFile(new URL("../src/events/projectors/qscore-projector.ts", import.meta.url), "utf8");
assert.equal(projector.includes("extractQscoreSignals"), false);
assert.equal(projector.includes("signalId"), false);
console.log("signal registry ownership test passed");
+16 -20
View File
@@ -4,12 +4,11 @@ import { db } from "../../db/client.js";
import {
growActiveMissions,
growEvents,
growQscoreSignals,
missionActions,
} from "../../db/schema.js";
import { listActiveMissionsPg } from "../../grow/persistence.js";
import { listMissionActions } from "../../missions/actions.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService } from "../../services/qscore-proxy.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService, getQscoreLatestSignalsFromService } from "../../services/qscore-proxy.js";
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === "object" && !Array.isArray(value);
@@ -30,7 +29,6 @@ async function platformAnalytics() {
completedMissions,
totalActions,
doneActions,
qscoreSignalCount,
] = await Promise.all([
scalarCount(growEvents),
scalarCount(growEvents, eq(growEvents.category, "service")),
@@ -39,7 +37,6 @@ async function platformAnalytics() {
scalarCount(growActiveMissions, eq(growActiveMissions.status, "completed")),
scalarCount(missionActions),
scalarCount(missionActions, eq(missionActions.status, "done")),
scalarCount(growQscoreSignals),
]);
const serviceUsage = await db
@@ -65,29 +62,28 @@ async function platformAnalytics() {
completedMissions,
missionActions: totalActions,
completedActions: doneActions,
qscoreSignals: qscoreSignalCount,
},
serviceUsage,
};
}
async function userQscoreAnalytics(userId: string) {
const result = await getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID);
const [result, latestEvidence] = await Promise.all([
getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID),
getQscoreLatestSignalsFromService(userId, DEFAULT_QSCORE_ORG_ID),
]);
const breakdown = result?.breakdown ?? {};
const latestSignals = Array.isArray(breakdown.signals) ? breakdown.signals : [];
const timelineRaw = Array.isArray(breakdown.signalTimeline) ? breakdown.signalTimeline : latestSignals;
const signalTimeline = timelineRaw.map((item) => {
const s = isRecord(item) ? item : {};
return {
signalId: typeof s.signalId === "string" ? s.signalId : typeof s.signal_id === "string" ? s.signal_id : "",
score: typeof s.score === "number" ? s.score : 0,
present: typeof s.present === "boolean" ? s.present : true,
source: typeof s.source === "string" ? s.source : "",
occurredAt: typeof s.occurredAt === "string" ? s.occurredAt : result?.calculated_at ?? new Date().toISOString(),
updatedAt: typeof s.updatedAt === "string" ? s.updatedAt : result?.calculated_at ?? new Date().toISOString(),
};
});
const latestSignals = latestEvidence ?? [];
const timelineRaw = latestSignals;
const signalTimeline = timelineRaw.map((signal) => ({
signalId: signal.signal_id,
score: signal.score,
present: signal.present,
source: signal.source,
occurredAt: signal.last_occurred_at,
updatedAt: signal.last_seen_at || null,
}));
return {
kind: "user-qscore",
@@ -101,7 +97,7 @@ async function userQscoreAnalytics(userId: string) {
sq_score: result.sq_score,
signalCount: signalTimeline.length,
dimensions: isRecord(breakdown.dimensions) ? breakdown.dimensions : result.quotients,
summary: typeof breakdown.summary === "string" ? breakdown.summary : null,
summary: null,
updatedAt: result.calculated_at || new Date().toISOString(),
}
: null,
+1 -1
View File
@@ -113,6 +113,7 @@ export const config = {
process.env.ROLEPLAY_PUBLIC_URL ?? process.env.ROLEPLAY_SERVICE_URL ?? "http://localhost:8008",
qscoreServiceUrl:
process.env.QSCORE_SERVICE_URL ?? "http://localhost:8000",
qscoreEventDeliveryTimeoutMs: Number(process.env.QSCORE_EVENT_DELIVERY_TIMEOUT_MS ?? 5_000),
resumeServiceUrl:
process.env.RESUME_SERVICE_URL ?? "http://localhost:8002",
userServiceUrl:
@@ -197,4 +198,3 @@ export const config = {
required, // exported so other modules can fail fast on boot
} as const;
+25
View File
@@ -59,10 +59,35 @@ async function ensureOnboardingTable() {
await db.execute(`CREATE UNIQUE INDEX IF NOT EXISTS onboarding_user_idx ON onboarding (user_id)`);
}
async function ensureEventOutboxTable() {
await db.execute(`
CREATE TABLE IF NOT EXISTS event_outbox (
id text PRIMARY KEY DEFAULT gen_random_uuid()::text NOT NULL,
topic text NOT NULL,
aggregate_id text NOT NULL,
destination text NOT NULL,
payload jsonb NOT NULL,
headers jsonb NOT NULL DEFAULT '{}'::jsonb,
status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'processing', 'published')),
attempts integer NOT NULL DEFAULT 0,
available_at timestamp with time zone NOT NULL DEFAULT now(),
locked_at timestamp with time zone,
published_at timestamp with time zone,
last_error text,
response jsonb,
created_at timestamp with time zone NOT NULL DEFAULT now(),
updated_at timestamp with time zone NOT NULL DEFAULT now()
)
`);
await db.execute(`CREATE INDEX IF NOT EXISTS event_outbox_delivery_idx ON event_outbox (status, available_at)`);
await db.execute(`CREATE UNIQUE INDEX IF NOT EXISTS event_outbox_topic_aggregate_idx ON event_outbox (topic, aggregate_id)`);
}
export async function ensureRuntimeSchema() {
await ensureUserPlanColumn();
await ensureGrowConversationsMetadataColumn();
await ensureSystemNotificationsTables();
await ensureOnboardingTable();
await ensureEventOutboxTable();
log.info("runtime schema ensured");
}
+30
View File
@@ -380,6 +380,36 @@ export const growEvents = pgTable(
}),
);
// Transactional delivery queue for canonical GrowEvents. The grow_events row
// and its QScore delivery intent are inserted in the same database transaction;
// delivery is retried independently and is idempotent on (topic, aggregateId).
export const eventOutbox = pgTable(
"event_outbox",
{
id: text("id").primaryKey().default(sql`gen_random_uuid()::text`),
topic: text("topic").notNull(),
aggregateId: text("aggregate_id").notNull(),
destination: text("destination").notNull(),
payload: jsonb("payload").$type<Record<string, unknown>>().notNull(),
headers: jsonb("headers").$type<Record<string, string>>().notNull().default(sql`'{}'::jsonb`),
status: text("status", {
enum: ["pending", "processing", "published"],
}).notNull().default("pending"),
attempts: integer("attempts").notNull().default(0),
availableAt: timestamp("available_at", { withTimezone: true }).defaultNow().notNull(),
lockedAt: timestamp("locked_at", { withTimezone: true }),
publishedAt: timestamp("published_at", { withTimezone: true }),
lastError: text("last_error"),
response: jsonb("response").$type<Record<string, unknown>>(),
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
},
(t) => ({
deliveryIdx: index("event_outbox_delivery_idx").on(t.status, t.availableAt),
aggregateIdx: uniqueIndex("event_outbox_topic_aggregate_idx").on(t.topic, t.aggregateId),
}),
);
export const missionServiceSessions = pgTable(
"mission_service_sessions",
{
+4 -2
View File
@@ -53,8 +53,10 @@ export type StoredGrowEvent = GrowEventEnvelope & {
export type QscoreSignal = {
signalId: string;
score: number;
present: boolean;
// QScore owns normalization and scoring. Raw readiness evidence returned by
// its event-ingest API may identify a projected signal before a score run.
score?: number;
present?: boolean;
source?: string;
raw?: Record<string, unknown>;
};
+1 -10
View File
@@ -8,7 +8,6 @@ import {
markGrowEventProcessing,
recordGrowEvent,
} from "./record-grow-event.js";
import { ensureOnboardingBaselineQscoreForCompletedAt } from "./onboarding-qscore.js";
import {
onboardingCompletedAtFromEvent,
runCuratorOnboardingLoopSafely,
@@ -120,12 +119,6 @@ export async function getLatestValidOnboardingLedgerEvent(userId: string): Promi
return statusEvent;
}
export async function ensureOnboardingBaselineQscoreFromLedger(userId: string) {
const event = await getLatestValidOnboardingLedgerEvent(userId);
if (!event) return false;
return ensureOnboardingBaselineQscoreForCompletedAt(userId, event.occurredAt);
}
function onboardingContextFromInput(context?: Record<string, unknown>) {
const input = context ?? {};
const preferences = asRecord(input.preferences);
@@ -140,7 +133,6 @@ function onboardingContextFromInput(context?: Record<string, unknown>) {
export async function ensureOnboardingSideEffectsForEvent(event: GrowEventRow, contextOverride?: Record<string, unknown>) {
if (!event.userId || !isValidOnboardingLedgerEvent(event)) {
return {
qscoreBaselineSeeded: false,
curatorOnboarding: { status: "skipped" as const, reason: event.userId ? "not_onboarding_completion" : "missing_user_id" },
missions: { status: "skipped" as const, reason: event.userId ? "not_onboarding_completion" : "missing_user_id", started: [], existing: [] },
};
@@ -151,7 +143,6 @@ export async function ensureOnboardingSideEffectsForEvent(event: GrowEventRow, c
completedAtFromOnboardingPayload(event.payload) ??
event.occurredAt.toISOString();
const qscoreBaselineSeeded = await ensureOnboardingBaselineQscoreForCompletedAt(event.userId, completedAt);
const curatorOnboarding = await runCuratorOnboardingLoopSafely({
userId: event.userId,
completedAt,
@@ -166,7 +157,7 @@ export async function ensureOnboardingSideEffectsForEvent(event: GrowEventRow, c
existing: [],
};
return { qscoreBaselineSeeded, curatorOnboarding, missions };
return { curatorOnboarding, missions };
}
export async function recordAndProcessOnboardingCompletion(input: {
-75
View File
@@ -1,75 +0,0 @@
import { forwardSignalsToQscoreService } from "../services/service-agents.js";
import { DEFAULT_QSCORE_ORG_ID } from "../services/qscore-proxy.js";
export const ONBOARDING_BASELINE_SIGNAL_ID = "onboarding.completed";
/**
* Workbook threshold for a completed onboarding baseline signal
* (Complete=Yes → 0.5). qscore_service owns the aggregate score; this is the
* per-signal evidence weight forwarded to it.
*/
export const ONBOARDING_BASELINE_SIGNAL_SCORE = 0.5;
function asRecord(value: unknown): Record<string, unknown> {
return value && typeof value === "object" && !Array.isArray(value) ? (value as Record<string, unknown>) : {};
}
function parseCompletedAt(value: unknown): Date | null {
if (value instanceof Date) {
return Number.isNaN(value.getTime()) ? null : value;
}
if (typeof value !== "string" || !value.trim()) return null;
const parsed = new Date(value);
return Number.isNaN(parsed.getTime()) ? new Date() : parsed;
}
function onboardingCompletedAt(preferences: Record<string, unknown> | undefined): Date | null {
const onboarding = asRecord(preferences?.onboarding);
return parseCompletedAt(onboarding.completed_at ?? onboarding.completedAt);
}
/**
* Forward an onboarding.completed baseline signal to qscore_service when
* onboarding is completed.
*
* qscore_service OWNS all scoring. The backend no longer seeds local qscore
* tables; instead it forwards a single baseline readiness signal and lets
* qscore_service compute the score. Idempotency is delegated to qscore_service's
* latest-signal upsert semantics.
*/
export async function ensureOnboardingBaselineQscore(
userId: string,
preferences: Record<string, unknown> | undefined,
): Promise<boolean> {
const completedAt = onboardingCompletedAt(preferences);
if (!completedAt) return false;
return ensureOnboardingBaselineQscoreForCompletedAt(userId, completedAt);
}
export async function ensureOnboardingBaselineQscoreForCompletedAt(
userId: string,
completedAtInput: string | Date,
): Promise<boolean> {
const completedAt = parseCompletedAt(completedAtInput);
if (!completedAt) return false;
await forwardSignalsToQscoreService({
orgId: DEFAULT_QSCORE_ORG_ID,
userId,
profession: "student",
source: "onboarding",
signals: [
{
signalId: ONBOARDING_BASELINE_SIGNAL_ID,
score: ONBOARDING_BASELINE_SIGNAL_SCORE,
present: true,
raw: {
reason: "completed onboarding baseline",
completedAt: completedAt.toISOString(),
},
},
],
});
return true;
}
+18 -382
View File
@@ -1,385 +1,21 @@
import { type GrowEventRow } from "../../db/schema.js";
import { asRecord, clampScore, getNumber, getString, type QscoreSignal } from "../envelope.js";
import { forwardSignalsToQscoreService } from "../../services/service-agents.js";
import { DEFAULT_QSCORE_ORG_ID } from "../../services/qscore-proxy.js";
import type { GrowEventRow } from "../../db/schema.js";
import { publishQscoreOutboxForEvent } from "../qscore-event-outbox.js";
function signal(signalId: string, score: number, raw?: Record<string, unknown>, present = true): QscoreSignal {
return { signalId, score: clampScore(score), present, raw };
}
function nestedNumber(record: Record<string, unknown>, keys: string[]): number | undefined {
for (const key of keys) {
const direct = getNumber(record[key]);
if (direct !== undefined) return direct;
}
return undefined;
}
const RESUME_UPLOAD_BASELINE_SCORE = 35;
function extractResumeSignals(event: GrowEventRow): QscoreSignal[] {
const payload = event.payload ?? {};
const analysis = asRecord(payload.analysis ?? payload.result ?? payload);
const scoreBreakdown = Array.isArray(analysis.score_breakdown) ? analysis.score_breakdown : [];
const dimensions = Array.isArray(analysis.dimensional_scores) ? analysis.dimensional_scores : [];
const byCategory = new Map<string, number>();
for (const item of scoreBreakdown) {
const row = asRecord(item);
const category = typeof row.category === "string" ? row.category : undefined;
const score = getNumber(row.score);
if (category && score !== undefined) byCategory.set(category, score);
}
const byDimension = new Map<string, number>();
for (const item of dimensions) {
const row = asRecord(item);
const dimension = typeof row.dimension === "string" ? row.dimension : undefined;
const score = getNumber(row.score);
if (dimension && score !== undefined) byDimension.set(dimension, score);
}
const signals: QscoreSignal[] = [];
if (event.type.includes("uploaded") || event.type.includes("created")) {
// Uploading a resume is only a baseline readiness signal. The actual Q Score
// should rise from parsed resume/interview/roleplay evidence, not jump to 100
// immediately after onboarding.
signals.push(signal("resume.uploaded", RESUME_UPLOAD_BASELINE_SCORE, { eventId: event.id }));
}
const ats = byCategory.get("ATS Compatibility") ?? nestedNumber(analysis, ["ats_score", "ats_compatibility", "atsCompatibility"]);
if (ats !== undefined) signals.push(signal("resume.ats_compatibility", ats, { eventId: event.id }));
const keywords = byDimension.get("Keywords") ?? nestedNumber(analysis, ["keyword_score", "keyword_relevance", "keywords"]);
if (keywords !== undefined) {
signals.push(signal("resume.keyword_relevance", keywords, { eventId: event.id }));
}
const quantification = byDimension.get("Quantification") ?? nestedNumber(analysis, ["quantification_score"]);
if (quantification !== undefined) signals.push(signal("resume.quantified_achievements", quantification, { eventId: event.id }));
const contentQuality = byCategory.get("Content Quality") ?? nestedNumber(analysis, ["content_quality", "clarity_score"]);
if (contentQuality !== undefined) signals.push(signal("resume.grammar_clarity", contentQuality, { eventId: event.id }));
const formatting = byCategory.get("Formatting") ?? nestedNumber(analysis, ["formatting", "format_score"]);
if (formatting !== undefined) signals.push(signal("resume.format_structure", formatting, { eventId: event.id }));
return signals;
}
function interviewSessionsScore(count: number): number {
if (count <= 0) return 0;
if (count <= 2) return 20;
if (count <= 5) return 35;
return 50;
}
function interviewDiversityScore(count: number): number {
if (count <= 1) return 10;
if (count === 2) return 20;
return 30;
}
function roleplayScenariosScore(count: number): number {
if (count <= 0) return 0;
if (count <= 3) return 20;
if (count <= 6) return 35;
return 50;
}
function extractInterviewSignals(event: GrowEventRow): QscoreSignal[] {
const payload = event.payload ?? {};
const review = asRecord(payload.review ?? payload.result ?? payload);
const status = String(review.status ?? payload.status ?? "");
if (!event.type.includes("review") && !event.type.includes("completed") && status !== "completed") return [];
const signals: QscoreSignal[] = [];
// A completed interview must always emit at least a single-completion signal.
// When session_count is absent, default to 1 so the event is not silently lost.
const sessionCount = getNumber(payload.session_count ?? payload.sessionCount) ?? 1;
signals.push(signal("interview.sessions_completed", interviewSessionsScore(sessionCount), { eventId: event.id, sessionCount }));
const typeDiversity = getNumber(payload.type_diversity ?? payload.typeDiversity);
if (typeDiversity !== undefined) {
signals.push(signal("interview.type_diversity", interviewDiversityScore(typeDiversity), { eventId: event.id, typeDiversity }));
}
const overall = getNumber(review.overall_score ?? review.overallScore ?? payload.overall_score);
if (overall !== undefined) signals.push(signal("interview.overall_score", overall, { eventId: event.id }));
const rubric = asRecord(review.rubric_scores ?? review.rubricScores);
const content = getNumber(rubric.content_quality ?? rubric.content ?? rubric.communication ?? review.communication_score);
if (content !== undefined) signals.push(signal("interview.response_clarity", content, { eventId: event.id }));
const roleAlignment = getNumber(rubric.role_alignment ?? rubric.technical_accuracy ?? review.role_alignment_score);
if (roleAlignment !== undefined) signals.push(signal("interview.technical_accuracy", roleAlignment, { eventId: event.id }));
const language = getNumber(rubric.language ?? review.language_score);
if (language !== undefined) signals.push(signal("interview.behavioral_quality", language, { eventId: event.id }));
const historical = asRecord(review.historical_comparison ?? review.historicalComparison);
const delta = getNumber(historical.overall_delta ?? historical.overallDelta);
if (delta !== undefined) signals.push(signal("interview.improvement_over_time", 50 + delta * 2.5, { eventId: event.id, delta }));
return signals;
}
function extractRoleplaySignals(event: GrowEventRow): QscoreSignal[] {
const payload = event.payload ?? {};
const review = asRecord(payload.review ?? payload.result ?? payload);
const status = String(review.status ?? payload.status ?? "");
if (!event.type.includes("review") && !event.type.includes("completed") && status !== "completed") return [];
const signals: QscoreSignal[] = [];
// A completed roleplay must always emit at least a single-completion signal.
// When scenario_count is absent, default to 1 so the event is not silently lost.
const scenarioCount = getNumber(payload.scenario_count ?? payload.scenarioCount) ?? 1;
signals.push(signal("roleplay.scenarios_completed", roleplayScenariosScore(scenarioCount), { eventId: event.id, scenarioCount }));
const rubric = asRecord(review.rubric_scores ?? review.rubricScores);
const scenario = getNumber(rubric.scenario_adherence ?? review.scenario_adherence_score);
if (scenario !== undefined) signals.push(signal("roleplay.situational_judgment", scenario, { eventId: event.id }));
const empathy = getNumber(rubric.emotional_intelligence ?? review.emotional_intelligence_score);
if (empathy !== undefined) signals.push(signal("roleplay.empathy_demonstrated", empathy, { eventId: event.id }));
const adaptability = getNumber(rubric.adaptability ?? review.adaptability_score);
if (adaptability !== undefined) signals.push(signal("roleplay.problem_resolution", adaptability, { eventId: event.id }));
const communication = getNumber(rubric.content ?? rubric.communication ?? review.communication_score);
if (communication !== undefined) signals.push(signal("roleplay.communication_effectiveness", communication, { eventId: event.id }));
return signals;
}
function coursesCompletedScore(count: number): number {
if (count <= 0) return 0;
if (count <= 3) return 35;
if (count <= 10) return 60;
return 80;
}
function courseCompletionRateScore(pct: number): number {
if (pct < 50) return 20;
if (pct <= 80) return 40;
return 60;
}
function courseDifficultyScore(difficulty: string): number | undefined {
const d = difficulty.toLowerCase();
if (d.includes("beginner") || d === "easy") return 20;
if (d.includes("inter") || d.includes("medium")) return 35;
if (d.includes("adv")) return 50;
return undefined;
}
function coursePathwayRelevanceScore(label: string): number | undefined {
const l = label.toLowerCase();
if (l.includes("high")) return 40;
if (l.includes("med")) return 25;
if (l.includes("low")) return 10;
return undefined;
}
function extractCourseSignals(event: GrowEventRow): QscoreSignal[] {
const payload = event.payload ?? {};
if (
!event.type.includes("progress_recorded") &&
!event.type.includes("completed") &&
!event.type.includes("started") &&
!event.type.includes("watch")
)
return [];
const watchPctRaw = getNumber(payload.watchPct ?? payload.watch_pct) ?? 0;
const watchPctScaled = watchPctRaw <= 1 ? watchPctRaw * 100 : watchPctRaw;
const completedCount = getNumber(payload.completed_count ?? payload.completedCount ?? payload.courses_completed);
const difficulty = getString(payload.difficulty);
const label = getString(payload.label ?? payload.pathway_label);
const raw = {
eventId: event.id,
courseId: payload.courseId ?? payload.course_id,
lessonId: payload.lessonId ?? payload.lesson_id,
watchPct: watchPctRaw,
/**
* Compatibility adapter for event/mission processing.
*
* The backend never extracts, normalizes, thresholds, or scores a signal. The
* canonical GrowEvent was queued transactionally when it was recorded. This
* method only attempts that delivery and passes QScore-owned projected evidence
* back to mission reducers when the QScore API returns it.
*/
export async function applyQscoreProjection(event: Pick<GrowEventRow, "id" | "userId">) {
if (!event.userId) return { signals: [], score: undefined, delivery: "skipped" as const };
const result = await publishQscoreOutboxForEvent(event.id);
return {
signals: result.signals,
score: undefined,
delivery: result.status,
response: result.response,
};
const signals: QscoreSignal[] = [];
signals.push(signal("courses.started", 40, raw));
signals.push(signal("courses.completion_rate", courseCompletionRateScore(watchPctScaled), { ...raw, completionPct: watchPctScaled }));
if (watchPctScaled >= 80) {
signals.push(signal("courses.completed", coursesCompletedScore(completedCount ?? 1), { ...raw, completedCount: completedCount ?? 1 }));
}
if (difficulty) {
const diffScore = courseDifficultyScore(difficulty);
if (diffScore !== undefined) signals.push(signal("courses.difficulty", diffScore, { ...raw, difficulty }));
}
if (label) {
const relScore = coursePathwayRelevanceScore(label);
if (relScore !== undefined) signals.push(signal("courses.pathway_relevance", relScore, { ...raw, label }));
}
return signals;
}
function sourceSignalPrefix(source: string) {
return source
.toLowerCase()
.replace(/-service$/, "")
.replace(/[^a-z0-9]+/g, "_")
.replace(/^_+|_+$/g, "") || "service";
}
function extractScoredServiceSignals(event: GrowEventRow): QscoreSignal[] {
const payload = event.payload ?? {};
const review = asRecord(payload.review ?? payload.result ?? payload);
const status = String(review.status ?? payload.status ?? "");
const isCompletion =
event.type.includes("completed") ||
event.type.includes("updated") ||
event.type.includes("signal_projected") ||
event.type.includes("signal.projected") ||
status === "completed";
if (!isCompletion) return [];
const score = getNumber(
payload.score ??
payload.qscore ??
payload.q_score ??
payload.readiness_score ??
payload.overall_score ??
review.score ??
review.qscore ??
review.q_score ??
review.readiness_score ??
review.overall_score,
);
if (score === undefined) return [];
const prefix = sourceSignalPrefix(event.source);
return [
signal(`${prefix}.service_completion_score`, score, {
eventId: event.id,
source: event.source,
type: event.type,
}),
];
}
function countFromPayload(payload: Record<string, unknown>): number | undefined {
const result = asRecord(payload.result);
const request = asRecord(payload.request);
const params = asRecord(request.params);
const direct = getNumber(payload.matchCount ?? payload.matches ?? payload.shortlisted ?? result.matchCount ?? result.matches ?? result.shortlisted);
if (direct !== undefined) return direct;
const opportunities = Array.isArray(result.opportunities) ? result.opportunities : undefined;
if (opportunities) return opportunities.length;
const requestOpportunities = Array.isArray(params.opportunities) ? params.opportunities : undefined;
return requestOpportunities?.length;
}
function jobsViewedScore(count: number): number {
if (count <= 0) return 0;
if (count <= 20) return 15;
if (count <= 50) return 30;
return 40;
}
function applicationsSubmittedScore(count: number): number {
if (count <= 0) return 0;
if (count <= 5) return 20;
if (count <= 15) return 35;
return 50;
}
function matchRateScore(pct: number): number {
if (pct < 50) return 15;
if (pct <= 75) return 35;
return 50;
}
function extractMatchmakingSignals(event: GrowEventRow): QscoreSignal[] {
if (!event.type.startsWith("matchmaking.")) return [];
const payload = event.payload ?? {};
const count = countFromPayload(payload);
const raw = {
eventId: event.id,
type: event.type,
action: payload.action,
opportunityId: payload.opportunityId ?? payload.matchId ?? payload.jobId ?? event.subject?.externalId,
taskId: payload.taskId ?? payload.curatorTaskId ?? event.correlation?.taskId,
matchCount: count,
status: payload.status,
};
const signals: QscoreSignal[] = [];
// Jobs viewed — from feed.viewed and match.viewed events with cumulative counts
if (event.type === "matchmaking.feed.viewed" || event.type === "matchmaking.match.viewed") {
const viewedCount = getNumber(payload.jobs_viewed ?? payload.jobsViewed ?? payload.viewed_count ?? payload.viewedCount ?? count);
signals.push(signal("matching.jobs_viewed", jobsViewedScore(viewedCount ?? 1), { ...raw, viewedCount }));
}
// Applications submitted — from match.applied and application.completed events
if (event.type === "matchmaking.match.applied" || event.type === "matchmaking.application.completed") {
const appliedCount = getNumber(payload.applications_submitted ?? payload.applicationsSubmitted ?? payload.applied_count ?? payload.appliedCount ?? count);
signals.push(signal("matching.applications_submitted", applicationsSubmittedScore(appliedCount ?? 1), { ...raw, appliedCount }));
}
// Application quality — from match quality scores if available
const qualityScore = getNumber(payload.application_quality ?? payload.applicationQuality ?? payload.match_quality ?? payload.matchQuality);
if (qualityScore !== undefined) {
signals.push(signal("matching.application_quality", qualityScore, { ...raw, qualityScore }));
}
// Match rate — from matches.generated with match data
if (event.type === "matchmaking.matches.generated" && count !== undefined) {
const totalCandidates = getNumber(payload.total_candidates ?? payload.totalCandidates) ?? count;
const rate = totalCandidates > 0 ? (count / totalCandidates) * 100 : 0;
signals.push(signal("matching.match_rate", matchRateScore(rate), { ...raw, matchRate: rate }));
}
return signals;
}
function extractSocialSignals(event: GrowEventRow): QscoreSignal[] {
const payload = event.payload ?? {};
const signals: QscoreSignal[] = [];
// social-branding sends pre-computed qscore_signals array — forward as-is
const incoming = Array.isArray(payload.qscore_signals) ? payload.qscore_signals : Array.isArray(payload.qscoreSignals) ? payload.qscoreSignals : undefined;
if (incoming) {
for (const entry of incoming) {
const record = asRecord(entry);
const id = getString(record.signalId ?? record.signal_id ?? record.id);
const score = getNumber(record.score ?? record.value);
if (id !== undefined && score !== undefined) {
signals.push(signal(id, score, { eventId: event.id, source: event.source }));
}
}
return signals;
}
// Fall back to inline fields for backward compatibility
const inline = asRecord(payload.signals);
for (const [key, value] of Object.entries(inline)) {
const score = getNumber(value);
if (score !== undefined) signals.push(signal(key, score, { eventId: event.id, source: event.source }));
}
return signals;
}
export function extractQscoreSignals(event: GrowEventRow): QscoreSignal[] {
const source = event.source.toLowerCase();
if (source.includes("resume") || event.type.startsWith("resume.")) return extractResumeSignals(event);
if (source.includes("interview") || event.type.startsWith("interview.")) return extractInterviewSignals(event);
if (source.includes("roleplay") || event.type.startsWith("roleplay.")) return extractRoleplaySignals(event);
if (source.includes("course") || event.type.startsWith("course.")) return extractCourseSignals(event);
if (source.includes("matchmaking") || event.type.startsWith("matchmaking.")) return extractMatchmakingSignals(event);
if (source.includes("social") || source.includes("branding") || source.includes("linkedin")) return extractSocialSignals(event);
if (source.includes("qscore") || event.type.startsWith("qscore.")) return extractScoredServiceSignals(event);
const scoredServiceSignals = extractScoredServiceSignals(event);
if (scoredServiceSignals.length) return scoredServiceSignals;
return [];
}
export async function applyQscoreProjection(event: GrowEventRow) {
if (!event.userId) return { signals: [], score: undefined };
const signals = extractQscoreSignals(event);
if (!signals.length) return { signals, score: undefined };
// qscore_service OWNS all scoring. The backend only extracts signals and
// forwards them; ingest persists signals and marks the user dirty for async
// score computation. It does NOT return a score, so score stays undefined
// here — readers fetch the computed score via the proxy endpoints.
await forwardSignalsToQscoreService({
orgId: event.orgId ?? DEFAULT_QSCORE_ORG_ID,
userId: event.userId,
profession: "student",
source: event.source,
signals,
});
return { signals, score: undefined };
}
+197
View File
@@ -0,0 +1,197 @@
import { and, eq, lt, lte, or, sql } from "drizzle-orm";
import { config } from "../config.js";
import { db } from "../db/client.js";
import { eventOutbox } from "../db/schema.js";
import { log } from "../log.js";
import { asRecord, getNumber, getString, type QscoreSignal } from "./envelope.js";
export const QSCORE_GROW_EVENT_TOPIC = "qscore.grow_event.raw";
export const QSCORE_GROW_EVENT_DESTINATION = "/v1/events/ingest";
const LOCK_TIMEOUT_MS = 60_000;
const DEFAULT_POLL_MS = 2_000;
const DEFAULT_BATCH_SIZE = 25;
export type QscoreEventDeliveryResult = {
status: "queued" | "published";
signals: QscoreSignal[];
response?: Record<string, unknown>;
};
type Deliver = (input: {
destination: string;
payload: Record<string, unknown>;
headers: Record<string, string>;
}) => Promise<Record<string, unknown>>;
function signalFromUnknown(value: unknown): QscoreSignal | null {
const row = asRecord(value);
const signalId = getString(row.signalId ?? row.signal_id ?? row.id);
if (!signalId) return null;
const score = getNumber(row.score ?? row.normalizedScore ?? row.normalized_score);
const present = typeof row.present === "boolean" ? row.present : undefined;
const suppliedRaw = asRecord(row.raw ?? row.evidence);
const policyEvidence = compactRecord({
rawValue: row.rawValue ?? row.raw_value,
rawUnit: row.rawUnit ?? row.raw_unit,
band: row.band,
materialChange: row.materialChange ?? row.material_change,
});
return {
signalId,
...(score !== undefined ? { score } : {}),
...(present !== undefined ? { present } : {}),
...(getString(row.source) ? { source: getString(row.source) } : {}),
raw: Object.keys(suppliedRaw).length ? suppliedRaw : policyEvidence,
};
}
function compactRecord(value: Record<string, unknown>) {
return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined));
}
/** Read only QScore-owned projected evidence; never infer or score locally. */
export function projectedSignalsFromQscoreResponse(body: Record<string, unknown>): QscoreSignal[] {
const breakdown = asRecord(body.breakdown);
const candidates = Array.isArray(body.projectedSignals)
? body.projectedSignals
: Array.isArray(body.projected_signals)
? body.projected_signals
: Array.isArray(body.signals)
? body.signals
: Array.isArray(breakdown.signals)
? breakdown.signals
: [];
return candidates.map(signalFromUnknown).filter((signal): signal is QscoreSignal => signal !== null);
}
async function postToQscore(input: Parameters<Deliver>[0]): Promise<Record<string, unknown>> {
const response = await fetch(`${config.qscoreServiceUrl.replace(/\/$/, "")}${input.destination}`, {
method: "POST",
headers: {
...input.headers,
...(config.a2aAllowedKey ? { authorization: `Bearer ${config.a2aAllowedKey}` } : {}),
},
body: JSON.stringify(input.payload),
signal: AbortSignal.timeout(config.qscoreEventDeliveryTimeoutMs),
});
const text = await response.text();
let body: unknown = {};
if (text) {
try { body = JSON.parse(text); } catch { body = { detail: text }; }
}
if (!response.ok) {
throw new Error(`qscore raw-event ingest failed (${response.status}): ${text.slice(0, 500)}`);
}
return asRecord(body);
}
export async function publishQscoreOutboxItem(
outboxId: string,
deliver: Deliver = postToQscore,
): Promise<QscoreEventDeliveryResult> {
const now = new Date();
const staleBefore = new Date(now.getTime() - LOCK_TIMEOUT_MS);
const [claimed] = await db
.update(eventOutbox)
.set({
status: "processing",
attempts: sql`${eventOutbox.attempts} + 1`,
lockedAt: now,
updatedAt: now,
})
.where(and(
eq(eventOutbox.id, outboxId),
or(
and(eq(eventOutbox.status, "pending"), lte(eventOutbox.availableAt, now)),
and(eq(eventOutbox.status, "processing"), lt(eventOutbox.lockedAt, staleBefore)),
),
))
.returning();
if (!claimed) {
const [existing] = await db.select().from(eventOutbox).where(eq(eventOutbox.id, outboxId)).limit(1);
const response = existing?.response ?? undefined;
return {
status: existing?.status === "published" ? "published" : "queued",
signals: response ? projectedSignalsFromQscoreResponse(response) : [],
response,
};
}
try {
const response = await deliver({
destination: claimed.destination,
payload: claimed.payload,
headers: claimed.headers,
});
await db.update(eventOutbox).set({
status: "published",
response,
publishedAt: new Date(),
lockedAt: null,
lastError: null,
updatedAt: new Date(),
}).where(eq(eventOutbox.id, claimed.id));
return { status: "published", signals: projectedSignalsFromQscoreResponse(response), response };
} catch (error) {
const delaySeconds = Math.min(300, 2 ** Math.min(claimed.attempts, 8));
await db.update(eventOutbox).set({
status: "pending",
availableAt: new Date(Date.now() + delaySeconds * 1_000),
lockedAt: null,
lastError: error instanceof Error ? error.message : String(error),
updatedAt: new Date(),
}).where(eq(eventOutbox.id, claimed.id));
throw error;
}
}
export async function publishQscoreOutboxForEvent(eventId: string): Promise<QscoreEventDeliveryResult> {
const [row] = await db
.select({ id: eventOutbox.id, status: eventOutbox.status, response: eventOutbox.response })
.from(eventOutbox)
.where(and(eq(eventOutbox.topic, QSCORE_GROW_EVENT_TOPIC), eq(eventOutbox.aggregateId, eventId)))
.limit(1);
if (!row) return { status: "queued", signals: [] };
if (row.status === "published") {
return {
status: "published",
signals: row.response ? projectedSignalsFromQscoreResponse(row.response) : [],
response: row.response ?? undefined,
};
}
try {
return await publishQscoreOutboxItem(row.id);
} catch (error) {
log.warn({ error, eventId }, "QScore raw GrowEvent queued for retry");
return { status: "queued", signals: [] };
}
}
export async function flushQscoreEventOutbox(limit = DEFAULT_BATCH_SIZE): Promise<number> {
const now = new Date();
const staleBefore = new Date(now.getTime() - LOCK_TIMEOUT_MS);
const rows = await db
.select({ id: eventOutbox.id })
.from(eventOutbox)
.where(and(
eq(eventOutbox.topic, QSCORE_GROW_EVENT_TOPIC),
or(
and(eq(eventOutbox.status, "pending"), lte(eventOutbox.availableAt, now)),
and(eq(eventOutbox.status, "processing"), lt(eventOutbox.lockedAt, staleBefore)),
),
))
.limit(limit);
const settled = await Promise.allSettled(rows.map((row) => publishQscoreOutboxItem(row.id)));
return settled.filter((result) => result.status === "fulfilled").length;
}
export function startQscoreEventOutboxPublisher(pollMs = DEFAULT_POLL_MS) {
const timer = setInterval(() => {
flushQscoreEventOutbox().catch((error) => log.error({ error }, "QScore event outbox flush failed"));
}, pollMs);
timer.unref();
void flushQscoreEventOutbox().catch((error) => log.error({ error }, "initial QScore event outbox flush failed"));
return () => clearInterval(timer);
}
+49 -15
View File
@@ -1,15 +1,10 @@
import { and, eq } from "drizzle-orm";
import { db } from "../db/client.js";
import { growEvents, missionServiceSessions, users, type GrowEventRow } from "../db/schema.js";
import { eventOutbox, growEvents, missionServiceSessions, type GrowEventRow } from "../db/schema.js";
import { asRecord, getString, type GrowEventEnvelope } from "./envelope.js";
import { normalizeGrowEvent } from "./normalize.js";
async function ensureUser(userId: string) {
await db
.insert(users)
.values({ id: userId, email: `${userId}@service.local`, displayName: userId })
.onConflictDoNothing();
}
import { QSCORE_GROW_EVENT_DESTINATION, QSCORE_GROW_EVENT_TOPIC } from "./qscore-event-outbox.js";
import { DEFAULT_QSCORE_ORG_ID } from "../services/qscore-proxy.js";
async function resolveUserId(event: GrowEventEnvelope): Promise<string | undefined> {
if (event.userId) return event.userId;
@@ -44,14 +39,13 @@ export async function recordGrowEvent(input: unknown, overrides: { userId?: stri
export async function recordGrowEventWithResult(input: unknown, overrides: { userId?: string; source?: string } = {}): Promise<{ event: GrowEventRow; inserted: boolean }> {
const normalized = normalizeGrowEvent(input, overrides);
const resolvedUserId = await resolveUserId(normalized);
if (resolvedUserId) await ensureUser(resolvedUserId);
const processingStatus = resolvedUserId ? "pending" : "unresolved";
const values = {
id: normalized.id,
userId: resolvedUserId,
orgId: normalized.orgId,
orgId: normalized.orgId ?? DEFAULT_QSCORE_ORG_ID,
source: normalized.source,
type: normalized.type,
category: normalized.category,
@@ -79,11 +73,29 @@ export async function recordGrowEventWithResult(input: unknown, overrides: { use
//
// NULL dedupeKey never collides (Postgres treats NULLs as distinct under a
// unique index), so events with no dedupe key always insert fresh.
const insertedRows = await db
.insert(growEvents)
.values(values)
.onConflictDoNothing()
.returning();
const canonicalEvent = canonicalGrowEventEnvelope(normalized, resolvedUserId);
const insertedRows = await db.transaction(async (tx) => {
const rows = await tx
.insert(growEvents)
.values(values)
.onConflictDoNothing()
.returning();
const inserted = rows[0];
if (inserted?.userId) {
await tx.insert(eventOutbox).values({
topic: QSCORE_GROW_EVENT_TOPIC,
aggregateId: inserted.id,
destination: QSCORE_GROW_EVENT_DESTINATION,
payload: canonicalEvent,
headers: {
"content-type": "application/json",
"idempotency-key": inserted.id,
},
}).onConflictDoNothing();
}
return rows;
});
const inserted = insertedRows[0];
if (inserted) {
@@ -120,6 +132,28 @@ export async function recordGrowEventWithResult(input: unknown, overrides: { use
return { event: existing, inserted: false };
}
export function canonicalGrowEventEnvelope(normalized: GrowEventEnvelope, resolvedUserId?: string) {
return compactRecord({
id: normalized.id,
userId: resolvedUserId,
orgId: normalized.orgId ?? DEFAULT_QSCORE_ORG_ID,
source: normalized.source,
type: normalized.type,
category: normalized.category,
occurredAt: normalized.occurredAt,
mission: normalized.mission,
subject: normalized.subject,
correlation: normalized.correlation,
payload: normalized.payload,
raw: normalized.raw,
dedupeKey: normalized.dedupeKey,
});
}
function compactRecord(value: Record<string, unknown>): Record<string, unknown> {
return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined));
}
/**
* Routing gate: only newly inserted, user-resolved, pending events should be
* enqueued to the user-event actor. Replays / dedupe hits / unresolved events
+17 -29
View File
@@ -8,15 +8,13 @@ import {
missionArtifacts,
missionServiceSessions,
missionSuggestions,
qscoreSnapshots,
users,
type GrowHomeNotificationRow,
type NewGrowHomeNotification,
} from "../db/schema.js";
import { interviewService, resumeService, roleplayService } from "../services/product-service-clients.js";
import { buildServiceLink } from "../services/service-registry.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService } from "../services/qscore-proxy.js";
import { ensureOnboardingBaselineQscoreFromLedger } from "../events/onboarding-ledger.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService, getQscoreLatestSignalsFromService } from "../services/qscore-proxy.js";
import { listAvailableMissionDefinitions } from "../missions/registry.js";
import { listServiceCapabilities } from "../workflows/service-capabilities.js";
import { buildCuratorSprint } from "../v1/curator/curator-store.js";
@@ -219,7 +217,7 @@ function buildDayOneSeeds(): SeedNotification[] {
function buildDynamicSeeds(ctx: HomeContext): SeedNotification[] {
const seeds = buildDayOneSeeds().filter((seed) => seed.moduleId === "pathways" || seed.moduleId === "rewards");
const profile = profileFromPreferences(ctx.preferences);
const qscore = ctx.qscore?.score ?? Math.round(ctx.qscoreSignals.reduce((sum, s) => sum + s.score, 0) / Math.max(ctx.qscoreSignals.length, 1));
const qscore = ctx.qscore?.score;
const ats = latestScore(ctx.qscoreSignals, "resume.ats_compatibility");
const interviewOverall = latestScore(ctx.qscoreSignals, "interview.overall_score");
const roleplayComms = latestScore(ctx.qscoreSignals, "roleplay.communication_effectiveness");
@@ -278,7 +276,7 @@ function buildDynamicSeeds(ctx: HomeContext): SeedNotification[] {
});
}
if (ctx.qscore || ctx.qscoreSignals.length) {
if (qscore !== undefined) {
pushSeed(seeds, {
moduleId: "pathways",
title: qscore >= 80 ? "Review your best job matches" : "Find better-fit job matches",
@@ -382,16 +380,16 @@ function buildDynamicSeeds(ctx: HomeContext): SeedNotification[] {
async function collectContext(userId: string, input: { userProfile?: Record<string, unknown>; preferences?: Record<string, unknown> } = {}): Promise<HomeContext> {
const [user] = await db.select({ id: users.id, email: users.email, displayName: users.displayName }).from(users).where(eq(users.id, userId)).limit(1);
const qscoreResult = await getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID);
const breakdown = qscoreResult?.breakdown ?? {};
const breakdownSignals = Array.isArray(breakdown.signals) ? breakdown.signals : [];
const qscoreSignals = breakdownSignals.map((item) => {
const s = isRecord(item) ? item : {};
const [qscoreResult, latestQscoreSignals] = await Promise.all([
getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID),
getQscoreLatestSignalsFromService(userId, DEFAULT_QSCORE_ORG_ID),
]);
const qscoreSignals = (latestQscoreSignals ?? []).map((s) => {
return {
signalId: typeof s.signalId === "string" ? s.signalId : typeof s.signal_id === "string" ? s.signal_id : "",
score: typeof s.score === "number" ? s.score : 0,
source: typeof s.source === "string" ? s.source : null,
updatedAt: new Date(typeof s.updatedAt === "string" ? s.updatedAt : (qscoreResult?.calculated_at ?? new Date().toISOString())),
signalId: s.signal_id,
score: s.score,
source: s.source,
updatedAt: new Date(s.last_seen_at || s.last_occurred_at || qscoreResult?.calculated_at || 0),
};
});
const activeMissions = await db
@@ -453,9 +451,9 @@ async function collectContext(userId: string, input: { userProfile?: Record<stri
qscore: qscoreResult
? {
score: qscoreResult.q_score,
signalCount: typeof breakdown.signalCount === "number" ? breakdown.signalCount : qscoreSignals.length,
summary: typeof breakdown.summary === "string" ? breakdown.summary : null,
dimensions: isRecord(breakdown.dimensions) ? breakdown.dimensions : isRecord(qscoreResult.quotients) ? qscoreResult.quotients : null,
signalCount: qscoreSignals.length,
summary: null,
dimensions: isRecord(qscoreResult.quotients) ? qscoreResult.quotients : null,
}
: undefined,
qscoreSignals,
@@ -578,15 +576,7 @@ function buildModules(rows: GrowHomeNotificationRow[], ctx: HomeContext, mode: H
}
async function buildIdentity(ctx: HomeContext) {
const score = ctx.qscore?.score && ctx.qscore.score > 0 ? ctx.qscore.score : Math.round(ctx.qscoreSignals.reduce((sum, s) => sum + s.score, 0) / Math.max(ctx.qscoreSignals.length, 1)) || 47;
const [baselineSnapshot] = await db
.select({ score: qscoreSnapshots.score })
.from(qscoreSnapshots)
.where(and(eq(qscoreSnapshots.userId, ctx.user?.id ?? ""), eq(qscoreSnapshots.snapshotType, "baseline")))
.orderBy(desc(qscoreSnapshots.createdAt))
.limit(1);
const baseline = baselineSnapshot?.score ?? Math.max(35, score - 29);
const from = Math.max(baseline, Math.min(score, score - Math.min(17, Math.max(5, score - baseline))));
const score = ctx.qscore?.score ?? null;
const name = ctx.user?.displayName || ctx.user?.email?.split("@")[0] || "GrowQR User";
const completedArtifacts = ctx.artifacts.filter((a) => a.status === "ready" || a.status === "approved").length;
@@ -594,9 +584,8 @@ async function buildIdentity(ctx: HomeContext) {
name,
caption: "Your living QR · scan to view profile",
qrSrc: undefined,
qx: { from, to: score, baseline },
qx: { from: null, to: score, baseline: null },
glance: [
{ value: Math.max(0, score - baseline), label: "Growth" },
{ value: Math.max(0, ctx.qscore?.signalCount ?? ctx.qscoreSignals.length), label: "Signals" },
{ value: completedArtifacts || ctx.events.length, label: "Done" },
],
@@ -604,7 +593,6 @@ async function buildIdentity(ctx: HomeContext) {
}
export async function getHomeFeed(userId: string, opts: { refresh?: boolean; userProfile?: Record<string, unknown>; preferences?: Record<string, unknown> } = {}): Promise<HomeFeedResponse> {
await ensureOnboardingBaselineQscoreFromLedger(userId);
await buildCuratorSprint(userId);
const ctx = await collectContext(userId, { userProfile: opts.userProfile, preferences: opts.preferences });
const persisted = await readPersistedNotifications(userId);
+1 -1
View File
@@ -28,7 +28,7 @@ export type HomeIdentity = {
name: string;
caption: string;
qrSrc?: string;
qx: { from: number; to: number; baseline: number };
qx: { from: number | null; to: number | null; baseline: number | null };
glance: { value: number; label: string }[];
};
+2
View File
@@ -23,6 +23,7 @@ import { analyticsRoutes } from "./routes/analytics.js";
import { logRoutes } from "./routes/logs.js";
import { v1Routes } from "./v1/index.js";
import { startGrowEventsRedisConsumer } from "./events/redis-consumer.js";
import { startQscoreEventOutboxPublisher } from "./events/qscore-event-outbox.js";
import { startPassiveMissionReviewLoop } from "./missions/passive-runner.js";
import { db } from "./db/client.js";
import { ensureRuntimeSchema } from "./db/ensure-runtime-schema.js";
@@ -48,6 +49,7 @@ async function main() {
await reconcileOnBoot();
startGrowEventsRedisConsumer().catch((err) => log.error({ err }, "failed to start grow events redis consumer"));
startQscoreEventOutboxPublisher();
startPassiveMissionReviewLoop();
const app = new Hono();
+2 -2
View File
@@ -44,7 +44,7 @@ async function ingest(body: unknown, userId?: string, source?: string) {
processingStatus: event.processingStatus,
route,
qscore: { signals: [], score: undefined, idempotent: true },
onboarding: { qscoreBaselineSeeded: false, curatorOnboarding: { status: "already_processed" } },
onboarding: { curatorOnboarding: { status: "already_processed" } },
};
}
if (!event.userId) {
@@ -53,7 +53,7 @@ async function ingest(body: unknown, userId?: string, source?: string) {
processingStatus: event.processingStatus,
route,
qscore: { signals: [], score: undefined, skipped: "missing_user_id" },
onboarding: { qscoreBaselineSeeded: false, curatorOnboarding: { status: "skipped", reason: "missing_user_id" } },
onboarding: { curatorOnboarding: { status: "skipped", reason: "missing_user_id" } },
};
}
+25 -45
View File
@@ -10,11 +10,10 @@ import { db } from "../db/client.js";
import { events, growEvents, missionServiceSessions } from "../db/schema.js";
import { recordGrowEvent } from "../events/record-grow-event.js";
import { routeGrowEventToUserActor } from "../events/route-to-user-actor.js";
import { ensureOnboardingBaselineQscoreFromLedger } from "../events/onboarding-ledger.js";
import { getRequestUserPreferences, getRequestUserProfile } from "../services/user-context.js";
import { buildMatchmakingGatewayEvent } from "../services/matchmaking-events.js";
import { log } from "../log.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService } from "../services/qscore-proxy.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService, getQscoreLatestSignalsFromService } from "../services/qscore-proxy.js";
const LANDING_AGENTS = [
{ id: "resume", title: "Resume", agent: "Resume Strategist", description: "Resume proof, versions, parsing, analysis, and mission artifacts.", route: "/agents/resume" },
@@ -22,19 +21,6 @@ const LANDING_AGENTS = [
{ id: "roleplay", title: "Roleplay", agent: "Roleplay Coach", description: "Negotiation, promotion, transition, and communication drills.", route: "/agents/roleplay" },
] as const;
const DEFAULT_QSCORE = {
q_score: 76,
profession: "Student",
quotients: {
SQ: 76,
XQ: 74,
CQm: 78,
VQ: 72,
DQ: 75,
GQ: 80,
},
};
function getString(value: unknown): string | undefined {
return typeof value === "string" && value.trim() ? value.trim() : undefined;
}
@@ -680,7 +666,9 @@ async function buildPersonalizedRoleplayConfigurePayload(req: Request, body: Jso
user_id: String(rest.user_id ?? userId),
org_id: String(rest.org_id ?? "growqr"),
metadata,
qscore: (rest.qscore as JsonObject | undefined) ?? (isRecord(userContext.qscore) ? userContext.qscore : DEFAULT_QSCORE),
...((rest.qscore as JsonObject | undefined) ?? (isRecord(userContext.qscore) ? userContext.qscore : undefined)
? { qscore: (rest.qscore as JsonObject | undefined) ?? userContext.qscore }
: {}),
user_context: userContext,
};
}
@@ -807,16 +795,13 @@ export function serviceRoutes() {
app.get("/qscore/current", async (c) => {
const userId = c.get("userId");
try {
await ensureOnboardingBaselineQscoreFromLedger(userId);
} catch (err) {
log.warn({ err, userId }, "failed to seed onboarding Q Score baseline before current Q Score read");
}
const qscoreResult = await getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID);
const [qscoreResult, latestSignals] = await Promise.all([
getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID),
getQscoreLatestSignalsFromService(userId, DEFAULT_QSCORE_ORG_ID),
]);
const score = qscoreResult?.q_score ?? null;
const breakdown = qscoreResult?.breakdown ?? {};
const breakdownSignals = Array.isArray(breakdown.signals) ? breakdown.signals : [];
const evidence = latestSignals ?? [];
const quotients = qscoreResult?.quotients ?? {};
const dimensionFromKey = (id: string, value: unknown): {
@@ -845,32 +830,27 @@ export function serviceRoutes() {
iq_score: qscoreResult?.iq_score ?? null,
eq_score: qscoreResult?.eq_score ?? null,
sq_score: qscoreResult?.sq_score ?? null,
signalCount: breakdownSignals.length,
summary: typeof breakdown.summary === "string" ? breakdown.summary : `Readiness score computed from ${breakdownSignals.length} current signal${breakdownSignals.length === 1 ? "" : "s"}.`,
signalCount: evidence.length,
breakdown,
formulaVersion: qscoreResult?.formula_version ?? null,
formulaId: qscoreResult?.formula_id ?? null,
ledgerSeqFrom: qscoreResult?.ledger_seq_from ?? null,
ledgerSeqTo: qscoreResult?.ledger_seq_to ?? null,
dimensions,
updatedAt: qscoreResult?.calculated_at ?? null,
},
signals: breakdownSignals.map((signal) => {
const s = isRecord(signal) ? signal : {};
return {
signalId: typeof s.signalId === "string" ? s.signalId : typeof s.signal_id === "string" ? s.signal_id : "",
score: typeof s.score === "number" ? Math.round(s.score) : 0,
present: typeof s.present === "boolean" ? s.present : true,
source: typeof s.source === "string" ? s.source : "",
occurredAt: typeof s.occurredAt === "string" ? s.occurredAt : qscoreResult?.calculated_at ?? new Date().toISOString(),
updatedAt: typeof s.updatedAt === "string" ? s.updatedAt : qscoreResult?.calculated_at ?? new Date().toISOString(),
};
}),
signals: evidence.map((signal) => ({
signalId: signal.signal_id,
score: signal.score,
present: signal.present,
source: signal.source,
raw: signal.raw,
occurredAt: signal.last_occurred_at,
updatedAt: signal.last_seen_at || null,
formulaVersion: signal.formula_version,
})),
};
await recordGatewayEvent({
userId,
source: "qscore-service",
type: "qscore.review.opened",
payload: { score, signalCount: breakdownSignals.length, source: "services.qscore.current" },
correlation: { taskId: curatorTaskIdFromRequest(c.req.raw) },
}).catch((err) => log.warn({ err, userId }, "failed to record qscore review event"));
return c.json(response);
});
+53
View File
@@ -55,6 +55,17 @@ export type QscoreServiceResult = {
*/
export type QscoreProxyResult = QscoreServiceResult;
export type QscoreLatestSignal = {
signal_id: string;
source: string | null;
present: boolean;
score: number;
raw: unknown;
last_occurred_at: string | null;
last_seen_at: string;
formula_version: string;
};
const QSCORE_PROXY_TIMEOUT_MS = Number(
process.env.QSCORE_PROXY_TIMEOUT_MS ?? 3500,
);
@@ -138,3 +149,45 @@ export async function getQscoreFromService(
return null;
}
}
/** Fetch QScore's real latest evidence ledger; an unavailable service is null. */
export async function getQscoreLatestSignalsFromService(
userId: string,
orgId: string = DEFAULT_QSCORE_ORG_ID,
timeoutMs: number = QSCORE_PROXY_TIMEOUT_MS,
): Promise<QscoreLatestSignal[] | null> {
const qscoreUserId = toQscoreUserId(userId);
const url = `${config.qscoreServiceUrl.replace(/\/$/, "")}/v1/signals/${encodeURIComponent(qscoreUserId)}?org_id=${encodeURIComponent(orgId)}`;
try {
const res = await fetch(url, {
headers: {
"content-type": "application/json",
...(config.a2aAllowedKey ? { authorization: `Bearer ${config.a2aAllowedKey}` } : {}),
},
signal: AbortSignal.timeout(timeoutMs),
});
if (res.status === 404) return [];
if (!res.ok) {
log.warn({ status: res.status, userId, orgId }, "qscore_service latest signals returned non-success status");
return null;
}
const body: unknown = await res.json();
if (!isRecord(body) || !Array.isArray(body.signals)) throw new Error("qscore_service latest signals response is invalid");
return body.signals.flatMap((value): QscoreLatestSignal[] => {
if (!isRecord(value) || typeof value.signal_id !== "string" || typeof value.score !== "number") return [];
return [{
signal_id: value.signal_id,
source: typeof value.source === "string" ? value.source : null,
present: typeof value.present === "boolean" ? value.present : true,
score: value.score,
raw: value.raw,
last_occurred_at: typeof value.last_occurred_at === "string" ? value.last_occurred_at : null,
last_seen_at: typeof value.last_seen_at === "string" ? value.last_seen_at : "",
formula_version: typeof value.formula_version === "string" ? value.formula_version : "",
}];
});
} catch (err) {
log.warn({ err, userId, orgId }, "qscore_service latest signals proxy call failed");
return null;
}
}
+5 -104
View File
@@ -1,9 +1,7 @@
import { config } from "../config.js";
import { createHash } from "node:crypto";
import { buildServiceSessionPath } from "./service-registry.js";
import type { QscoreSignal } from "../events/envelope.js";
import { toQscoreUserId } from "./qscore-proxy.js";
import { serviceJson as productServiceJson } from "./product-service-clients.js";
import { getQscoreFromService, toQscoreUserId } from "./qscore-proxy.js";
// Lightweight agent reference (works with both old AgentProfile and new SubAgentModule).
export type ServiceAgentRef = {
@@ -129,6 +127,7 @@ async function runInterviewService(ctx: ServiceAgentContext): Promise<ServiceAge
}
async function runRoleplayService(ctx: ServiceAgentContext): Promise<ServiceAgentResult> {
const currentQscore = await getQscoreFromService(ctx.userId, ctx.orgId ?? "growqr");
const payload = {
user_id: ctx.userId,
org_id: ctx.orgId ?? "growqr",
@@ -141,16 +140,7 @@ async function runRoleplayService(ctx: ServiceAgentContext): Promise<ServiceAgen
difficulty: "medium",
source: "growqr-workflow",
},
qscore: {
q_score: 78,
profession: "student",
formula_version: "workflow-demo",
quotients: {
CQm: { score: 72, active: true },
XQ: { score: 80, active: true },
VQ: { score: 76, active: true },
},
},
...(currentQscore ? { qscore: currentQscore } : {}),
};
const detail = await serviceJson<Record<string, unknown>>(
config.roleplayServiceUrl,
@@ -174,55 +164,7 @@ async function runRoleplayService(ctx: ServiceAgentContext): Promise<ServiceAgen
async function runQScoreService(ctx: ServiceAgentContext): Promise<ServiceAgentResult> {
const orgId = ctx.orgId ?? "growqr";
const qscoreUserId = stableUuid(ctx.userId);
const signals = [
{
signal_id: "resume.uploaded",
present: true,
score: 82,
raw: { source: "resume-agent", workflow_goal: ctx.goal },
},
{
signal_id: "resume.ats_compatibility",
present: true,
score: 76,
raw: { source: "resume-agent", workflow_goal: ctx.goal },
},
{
signal_id: "engagement.features_used",
present: true,
score: 88,
raw: { source: "grow-agent", workflow_goal: ctx.goal },
},
{
signal_id: "goals.goal_clarity",
present: true,
score: 90,
raw: { source: "grow-agent", workflow_goal: ctx.goal },
},
];
// Try to ingest signals (non-critical — may fail if QScore worker is down)
let ingest: Record<string, unknown> | undefined;
try {
ingest = await serviceJson<Record<string, unknown>>(
config.qscoreServiceUrl,
"/v1/signals/ingest",
{
method: "POST",
body: JSON.stringify({
org_id: orgId,
user_id: qscoreUserId,
profession: "student",
source: "growqr-workflow",
signals,
}),
},
);
} catch (err) {
// Signal ingestion is optional — compute may still work with cached signals
ingest = { status: "skipped", reason: err instanceof Error ? err.message : String(err) };
}
const qscoreUserId = toQscoreUserId(ctx.userId);
// Try to compute Q-Score
let compute: Record<string, unknown> | undefined;
@@ -243,8 +185,6 @@ async function runQScoreService(ctx: ServiceAgentContext): Promise<ServiceAgentR
status: "unavailable",
summary: `Q Score compute failed; no score was generated: ${err instanceof Error ? err.message : String(err)}`,
detail: {
ingest,
signal_scores: signals.map(s => ({ id: s.signal_id, score: s.score })),
compute_error: err instanceof Error ? err.message : String(err),
},
};
@@ -253,49 +193,10 @@ async function runQScoreService(ctx: ServiceAgentContext): Promise<ServiceAgentR
return {
status: "ok",
summary: `Q Score computed Q Score ${compute.q_score ?? "(unknown)"} for ${ctx.goal}.`,
detail: { ingest, compute, qscore_user_id: qscoreUserId },
detail: { compute, qscore_user_id: qscoreUserId },
};
}
/**
* Forward extracted Grow signals to qscore_service for scoring.
*
* qscore_service OWNS all scoring; the backend only extracts signals and feeds
* them here. The userId is hashed to the qscore UUID via toQscoreUserId (from
* qscore-proxy.ts, the single source of truth shared with readers). The ingest
* endpoint persists signals and marks the user dirty for async score
* computation — it does NOT return a score, so the caller treats the result as
* fire-and-forget.
*/
export async function forwardSignalsToQscoreService(params: {
orgId: string;
userId: string;
profession: string;
source: string;
signals: QscoreSignal[];
}): Promise<void> {
const qscoreUserId = toQscoreUserId(params.userId);
await productServiceJson(
config.qscoreServiceUrl,
"/v1/signals/ingest",
{
method: "POST",
body: {
org_id: params.orgId,
user_id: qscoreUserId,
profession: params.profession,
source: params.source,
signals: params.signals.map((s) => ({
signal_id: s.signalId,
present: s.present,
score: s.score,
raw: s.raw ?? null,
})),
},
},
);
}
// ── Resume Building (resume-builder service from growqr-app) ──
async function runResumeAnalyze(ctx: ServiceAgentContext): Promise<ServiceAgentResult> {
+12 -13
View File
@@ -7,7 +7,7 @@ import { growEvents } from "../../db/schema.js";
import { recordGrowEvent } from "../../events/record-grow-event.js";
import { routeGrowEventToUserActor } from "../../events/route-to-user-actor.js";
import { v1AnalyticsActor } from "./analytics-actor.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService } from "../../services/qscore-proxy.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService, getQscoreLatestSignalsFromService } from "../../services/qscore-proxy.js";
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === "object" && !Array.isArray(value);
@@ -45,17 +45,16 @@ export function v1AnalyticsRoutes() {
app.get("/insight-snapshot", async (c) => {
const userId = c.get("userId");
const qscoreResult = await getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID);
const [qscoreResult, qscoreEvidence] = await Promise.all([
getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID),
getQscoreLatestSignalsFromService(userId, DEFAULT_QSCORE_ORG_ID),
]);
const breakdown = qscoreResult?.breakdown ?? {};
const breakdownSignals = Array.isArray(breakdown.signals) ? breakdown.signals : [];
const latestSignals = breakdownSignals.map((item) => {
const s = isRecord(item) ? item : {};
return {
signalId: typeof s.signalId === "string" ? s.signalId : typeof s.signal_id === "string" ? s.signal_id : "",
score: typeof s.score === "number" ? s.score : 0,
source: typeof s.source === "string" ? s.source : null,
};
});
const latestSignals = (qscoreEvidence ?? []).map((signal) => ({
signalId: signal.signal_id,
score: signal.score,
source: signal.source,
}));
const recentEvents = await db
.select()
.from(growEvents)
@@ -79,9 +78,9 @@ export function v1AnalyticsRoutes() {
const score = qscoreResult?.q_score ?? null;
const strongestSignal = [...latestSignals].sort((a, b) => b.score - a.score)[0];
const weakestSignal = [...latestSignals].sort((a, b) => a.score - b.score)[0];
const readinessSummary = typeof breakdown.summary === "string" ? breakdown.summary : "No projected readiness summary is available yet.";
const readinessSummary = null;
const lastUpdatedAt = qscoreResult?.calculated_at ?? null;
const signalCount = typeof breakdown.signalCount === "number" ? breakdown.signalCount : latestSignals.length;
const signalCount = latestSignals.length;
const response = {
roleFit: {
+6 -7
View File
@@ -2,7 +2,7 @@ import { and, desc, eq } from "drizzle-orm";
import { db } from "../../db/client.js";
import { growEvents, onboarding } from "../../db/schema.js";
import { asRecord, getString } from "../../events/envelope.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService } from "../../services/qscore-proxy.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService, getQscoreLatestSignalsFromService } from "../../services/qscore-proxy.js";
import { config } from "../../config.js";
import type { CuratorTask } from "./curator-types.js";
@@ -178,13 +178,12 @@ export async function buildCuratorUserContext(userId: string): Promise<CuratorUs
.orderBy(desc(growEvents.occurredAt))
.limit(80);
const [qscoreResult, resumeState, socialState] = await Promise.all([
const [qscoreResult, qscoreEvidence, resumeState, socialState] = await Promise.all([
getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID),
getQscoreLatestSignalsFromService(userId, DEFAULT_QSCORE_ORG_ID),
fetchProfileState(config.resumeServiceUrl, userId),
fetchProfileState(config.socialBrandingServiceUrl, userId),
]);
const breakdown = qscoreResult?.breakdown ?? {};
const breakdownSignals = Array.isArray(breakdown.signals) ? breakdown.signals : [];
const targetRole = fallbackCuratorRole(await resolveCuratorTargetRole({ userId })) ?? null;
const corpus = rows.map((row) => `${row.type} ${row.source} ${payloadText(row.payload)}`).join(" ").toLowerCase();
@@ -225,9 +224,9 @@ export async function buildCuratorUserContext(userId: string): Promise<CuratorUs
},
qscore: {
score: qscoreResult?.q_score ?? null,
signalCount: typeof breakdown.signalCount === "number" ? breakdown.signalCount : breakdownSignals.length,
summary: typeof breakdown.summary === "string" ? breakdown.summary : null,
dimensions: isRecord(breakdown.dimensions) ? breakdown.dimensions : (qscoreResult ? qscoreResult.quotients : null),
signalCount: qscoreEvidence?.length ?? 0,
summary: null,
dimensions: qscoreResult ? qscoreResult.quotients : null,
},
};
}
+39 -30
View File
@@ -1,15 +1,11 @@
import { Hono } from "hono";
import { requireUser, type AuthContext } from "../../auth/clerk.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService } from "../../services/qscore-proxy.js";
import { DEFAULT_QSCORE_ORG_ID, getQscoreFromService, getQscoreLatestSignalsFromService } from "../../services/qscore-proxy.js";
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === "object" && !Array.isArray(value);
}
function numberOr(value: unknown, fallback: number): number {
return typeof value === "number" && Number.isFinite(value) ? value : fallback;
}
function dimensionFromKey(id: string, value: unknown) {
const num =
typeof value === "number"
@@ -33,24 +29,37 @@ export function v1QscoreRoutes() {
app.get("/latest", async (c) => {
const userId = c.get("userId");
const result = await getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID);
const [result, latestSignals] = await Promise.all([
getQscoreFromService(userId, DEFAULT_QSCORE_ORG_ID),
getQscoreLatestSignalsFromService(userId, DEFAULT_QSCORE_ORG_ID),
]);
if (!result) {
const evidence = latestSignals ?? [];
return c.json({
status: "baseline_needed",
status: evidence.length ? "evidence_pending" : "baseline_needed",
score: null,
dimensions: [],
trendLabel: "No QScore signals yet",
trendLabel: evidence.length ? "Evidence received; score pending" : "No QScore signals yet",
lastUpdatedAt: null,
explanation: "No onboarding, service completion, or readiness signals have been projected for this user.",
signalCount: 0,
signals: [],
explanation: null,
signalCount: evidence.length,
signals: evidence.map((signal) => ({
signalId: signal.signal_id,
score: signal.score,
present: signal.present,
source: signal.source,
raw: signal.raw,
occurredAt: signal.last_occurred_at,
updatedAt: signal.last_seen_at || null,
formulaVersion: signal.formula_version,
})),
source: "qscore_service",
});
}
const breakdown = result.breakdown;
const breakdownSignals = Array.isArray(breakdown.signals) ? breakdown.signals : [];
const evidence = latestSignals ?? [];
const dimensions = Object.entries(result.quotients)
.map(([id, value]) => dimensionFromKey(id, value))
.filter((d): d is { id: string; label: string; score: number; signalCount: number; sources: string[] } => d !== null);
@@ -60,28 +69,28 @@ export function v1QscoreRoutes() {
status: "ready",
score: result.q_score,
dimensions,
trendLabel: breakdownSignals.length > 1 ? "Updated from recent activity" : "Baseline established",
trendLabel: evidence.length > 1 ? "Updated from recent activity" : "Evidence established",
lastUpdatedAt,
explanation:
typeof breakdown.summary === "string"
? breakdown.summary
: `Readiness score computed from ${breakdownSignals.length} current signal${breakdownSignals.length === 1 ? "" : "s"}.`,
signalCount: numberOr(breakdown.signalCount, breakdownSignals.length),
explanation: null,
signalCount: evidence.length,
iq_score: result.iq_score,
eq_score: result.eq_score,
sq_score: result.sq_score,
signals: breakdownSignals.map((signal) => {
const s = isRecord(signal) ? signal : {};
return {
signalId: typeof s.signalId === "string" ? s.signalId : typeof s.signal_id === "string" ? s.signal_id : "",
score: typeof s.score === "number" ? Math.round(s.score) : 0,
present: typeof s.present === "boolean" ? s.present : true,
source: typeof s.source === "string" ? s.source : "",
sourceEventId: typeof s.sourceEventId === "string" ? s.sourceEventId : typeof s.source_event_id === "string" ? s.source_event_id : null,
occurredAt: typeof s.occurredAt === "string" ? s.occurredAt : result.calculated_at,
updatedAt: typeof s.updatedAt === "string" ? s.updatedAt : result.calculated_at,
};
}),
formulaVersion: result.formula_version,
formulaId: result.formula_id,
ledgerSeqFrom: result.ledger_seq_from,
ledgerSeqTo: result.ledger_seq_to,
breakdown,
signals: evidence.map((signal) => ({
signalId: signal.signal_id,
score: signal.score,
present: signal.present,
source: signal.source,
raw: signal.raw,
occurredAt: signal.last_occurred_at,
updatedAt: signal.last_seen_at || null,
formulaVersion: signal.formula_version,
})),
source: "qscore_service",
});
});