feat(reports): add birth time sensitivity

This commit is contained in:
Jesse_Chen
2026-09-04 02:11:57 +08:00
parent 44d3d3931b
commit a7041529c1
21 changed files with 1630 additions and 205 deletions
+37 -1
View File
@@ -164,7 +164,7 @@ export async function POST(request: Request) {
if (userId) {
const result = await supabase
.from("profiles")
.select("name,birth_date,reported_birth_time,active_birth_time,birth_time_source,birth_time_status,latitude,longitude,timezone_offset,birth_place_label,ayanamsa")
.select("name,birth_date,reported_birth_time,active_birth_time,birth_time_source,birth_time_status,rectification_case_id,declared_window_start,declared_window_end,uncertainty_before_minutes,uncertainty_after_minutes,latitude,longitude,timezone_offset,birth_place_label,ayanamsa")
.eq("id", userId)
.maybeSingle();
profile = result.data ?? null;
@@ -182,6 +182,42 @@ export async function POST(request: Request) {
userId,
rawBody: await request.json().catch(() => null),
profile,
loadCandidateRange: async () => {
const profileRow = profile && typeof profile === "object" ? profile as Record<string, unknown> : {};
const caseId = typeof profileRow.rectification_case_id === "string"
? profileRow.rectification_case_id
: null;
if (caseId) {
const { data, error } = await admin
.from("birth_time_rectification_cases")
.select("candidate_start,candidate_end")
.eq("id", caseId)
.eq("user_id", userId as string)
.in("status", ["confirmed", "completed"])
.maybeSingle();
if (error) throw error;
const row = data && typeof data === "object" ? data as Record<string, unknown> : null;
if (row && typeof row.candidate_start === "string" && typeof row.candidate_end === "string") {
return { startTime: row.candidate_start, endTime: row.candidate_end };
}
}
const { data, error } = await admin
.from("agentic_rectification_cases")
.select("candidate_range,updated_at")
.eq("user_id", userId as string)
.eq("status", "candidate_accepted")
.order("updated_at", { ascending: false })
.limit(1)
.maybeSingle();
if (error) throw error;
const row = data && typeof data === "object" ? data as Record<string, unknown> : null;
const range = row?.candidate_range && typeof row.candidate_range === "object"
? row.candidate_range as Record<string, unknown>
: null;
return range && typeof range.start_time === "string" && typeof range.end_time === "string"
? { startTime: range.start_time, endTime: range.end_time }
: null;
},
checkSessionOwned: async (sessionId) => {
const { data, error } = await supabase
.from("chat_sessions")
@@ -517,7 +517,8 @@ function EvidenceAppendixSection({ document }: { document: ReportDocument }) {
const appendix = document.evidenceAppendix;
const expandedByDefault = appendix.expandedByDefault || document.presentationMode === "research";
const hasContent = appendix.techniqueAudit.length > 0 || appendix.conflicts.length > 0
|| appendix.calculationEvidence.length > 0 || appendix.blockedTechniques.length > 0;
|| appendix.calculationEvidence.length > 0 || appendix.blockedTechniques.length > 0
|| Boolean(appendix.birthTimeSensitivity);
return (
<section aria-labelledby="report-appendix" className="personal-report-section personal-report-appendix">
@@ -556,6 +557,24 @@ function EvidenceAppendixSection({ document }: { document: ReportDocument }) {
{appendix.blockedTechniques.map((name, index) => <li key={`${name}-${index}`}>{name}</li>)}
</ul></>
)}
{appendix.birthTimeSensitivity && (
<section>
<h3></h3>
<p>{appendix.birthTimeSensitivity.window.startTime}{appendix.birthTimeSensitivity.window.endTime}{appendix.birthTimeSensitivity.window.representativeTime}</p>
<ul>
{appendix.birthTimeSensitivity.themes.map((theme) => (
<li key={theme.theme}>
<strong>{themeLabel(theme.theme)}</strong>{theme.status === "sensitive" ? "敏感" : "稳定"}
{theme.sensitiveLayers.length > 0 ? `(变化层:${theme.sensitiveLayers.join("、")}` : ""}
{theme.minuteVariations.map((variation) => (
<small key={variation.layer}>{variation.layer}{variation.values.map((entry) => `${entry.minute}=${entry.value}`).join("")}</small>
))}
</li>
))}
</ul>
<p>{appendix.birthTimeSensitivity.claimBoundary}</p>
</section>
)}
</div>
</details>
)}
@@ -141,12 +141,36 @@ const calculationEvidenceRowSchema = z.strictObject({
source: text(200),
});
const birthTimeSensitivitySchema = z.strictObject({
window: z.strictObject({
startTime: text(5),
endTime: text(5),
representativeTime: text(5),
candidateCount: z.number().int().min(2).max(15),
}),
themes: z.array(z.strictObject({
theme: themeIdSchema,
status: z.enum(["stable", "sensitive"]),
stableLayers: textArray(24, 100),
sensitiveLayers: textArray(24, 100),
minuteVariations: z.array(z.strictObject({
layer: text(100),
values: z.array(z.strictObject({
minute: z.string().regex(/^(?:[01]\d|2[0-3]):[0-5]\d$/),
value: text(300),
})).min(2).max(15),
})).max(24),
})).max(12),
claimBoundary: text(500),
});
const evidenceAppendixSchema = z.strictObject({
expandedByDefault: z.boolean(),
techniqueAudit: z.array(techniqueAuditRowSchema).max(100),
conflicts: z.array(conflictRowSchema).max(50),
calculationEvidence: z.array(calculationEvidenceRowSchema).max(100),
blockedTechniques: textArray(100, 120),
birthTimeSensitivity: birthTimeSensitivitySchema.optional(),
});
const subjectSchema = z.strictObject({
+78 -1
View File
@@ -29,6 +29,7 @@ import type {
ReportFunctionalRole,
ReportFunctionalRoleFact,
ReportInterpretiveFacts,
ReportBirthTimeSensitivityFact,
ReportPlanetaryFriendshipFact,
ReportSavScoreFact,
ReportShadbalaRankFact,
@@ -1343,6 +1344,52 @@ function readSavFacts(
};
}
function readBirthTimeSensitivity(workflow: JsonRecord): ReportBirthTimeSensitivityFact | undefined {
const packet = record(workflow.birth_time_sensitivity);
const window = record(packet?.window);
const rawThemes = record(packet?.theme_sensitivity);
const sensitiveLayers = record(record(packet?.sensitive_evidence)?.layers);
const candidateCount = finiteNumber(window?.candidate_count);
const startTime = text(window?.start_time);
const endTime = text(window?.end_time);
const representativeTime = text(window?.representative_time);
const claimBoundary = text(packet?.claim_boundary);
if (packet?.status !== "candidate_window_only" || !startTime || !endTime || !representativeTime
|| candidateCount === null || candidateCount < 2 || candidateCount > 15 || !claimBoundary) return undefined;
const themeAliases: Readonly<Record<string, string>> = { health: "health_pressure" };
const themes = Object.entries(rawThemes ?? {}).flatMap(([rawTheme, value]) => {
const row = record(value);
if (row?.status !== "stable" && row?.status !== "sensitive") return [];
const requestedSensitiveLayers = stringArray(row.sensitive_layers).slice(0, 24);
const minuteVariations = requestedSensitiveLayers.flatMap((layer) => {
const valuesByMinute = record(sensitiveLayers?.[layer]);
const values = Object.entries(valuesByMinute ?? {}).flatMap(([minute, rawValue]) => {
if (!/^(?:[01]\d|2[0-3]):[0-5]\d$/.test(minute)) return [];
const value = rawValue !== null && typeof rawValue === "object"
? canonicalSerialize(rawValue)
: String(rawValue ?? "");
const cleaned = value.replace(/[\u0000-\u001f\u007f]+/g, " ").replace(/\s+/g, " ").trim().slice(0, 300);
return cleaned && /[\p{L}\p{N}]/u.test(cleaned) ? [{ minute, value: cleaned }] : [];
}).sort((a, b) => a.minute.localeCompare(b.minute));
return values.length >= 2 && new Set(values.map((entry) => entry.value)).size >= 2
? [{ layer, values }]
: [];
}).sort((a, b) => a.layer.localeCompare(b.layer));
return [{
theme: themeAliases[rawTheme] ?? rawTheme,
status: minuteVariations.length > 0 ? "sensitive" as const : "stable" as const,
stableLayers: stringArray(row.stable_layers).slice(0, 24),
sensitiveLayers: minuteVariations.map((entry) => entry.layer),
minuteVariations,
}];
});
return {
window: { startTime, endTime, representativeTime, candidateCount },
themes,
claimBoundary: claimBoundary.slice(0, 500),
};
}
function readCurrentDasha(workflow: JsonRecord): ReportCurrentDashaFact | null {
const snapshot = readEvidenceSnapshot(workflow);
const modules = readChartModules(workflow);
@@ -1828,6 +1875,10 @@ export function buildReportEvidenceBundleV2(
}
return null;
})();
const birthTimeSensitivity = collectFirst((workflow) => {
const fact = readBirthTimeSensitivity(workflow);
return fact ? [fact] : [];
})[0];
const interpretiveFacts: ReportInterpretiveFacts = {
yogas: yogaRef ? collectFirst((workflow) => readYogaFacts(workflow, yogaRef)) : [],
functionalRoles: functionalRef
@@ -1840,6 +1891,7 @@ export function buildReportEvidenceBundleV2(
convergenceDomains: collectFirst(readConvergenceDomains),
planetaryFriendship,
pratyantarTimeline,
...(birthTimeSensitivity ? { birthTimeSensitivity } : {}),
};
// --- theme narrative seeds ----------------------------------------------
@@ -2559,6 +2611,21 @@ function evidenceAppendixFromBundle(
blockedTechniques: uniqueInOrder(bundle.blockedSections.flatMap((section) => (
section.missingTechniqueRefs.map((ref) => ledgerById.get(ref)?.technique ?? ref)
))).slice(0, 100),
...(bundle.interpretiveFacts.birthTimeSensitivity
? { birthTimeSensitivity: {
...bundle.interpretiveFacts.birthTimeSensitivity,
window: { ...bundle.interpretiveFacts.birthTimeSensitivity.window },
themes: bundle.interpretiveFacts.birthTimeSensitivity.themes.map((theme) => ({
...theme,
stableLayers: [...theme.stableLayers],
sensitiveLayers: [...theme.sensitiveLayers],
minuteVariations: theme.minuteVariations.map((variation) => ({
...variation,
values: variation.values.map((value) => ({ ...value })),
})),
})),
} }
: {}),
};
}
@@ -2611,7 +2678,11 @@ export function assembleReportDocumentV2(
id: section.id,
theme: section.theme,
title: section.title,
narrative: section.narrative,
narrative: bundle.interpretiveFacts.birthTimeSensitivity?.themes.some(
(theme) => theme.theme === section.theme && theme.status === "sensitive",
)
? `${section.narrative}\n\n本主题在当前出生时间可信区间内存在层级变化,以下内容只能作条件性解读,不能用于认定唯一出生分钟。`
: section.narrative,
actions: [...section.actions],
caveats: [...section.caveats],
claimStatus: section.claimStatus,
@@ -3215,6 +3286,12 @@ export function filterReportEvidenceBundleForSection(
convergenceDomains: source.interpretiveFacts.convergenceDomains,
planetaryFriendship: source.interpretiveFacts.planetaryFriendship,
pratyantarTimeline: section.theme === "timing" ? source.interpretiveFacts.pratyantarTimeline : null,
...(source.interpretiveFacts.birthTimeSensitivity ? {
birthTimeSensitivity: {
...source.interpretiveFacts.birthTimeSensitivity,
themes: source.interpretiveFacts.birthTimeSensitivity.themes.filter((theme) => theme.theme === section.theme),
},
} : {}),
};
const charts = source.charts.filter((chart) => (chart.id === "D1"
|| ledger.some((receipt) => receipt.technique.toUpperCase() === chart.id)));
@@ -191,6 +191,7 @@ export type ReportCreateCoreDeps = Readonly<{
userId: string | null;
rawBody: unknown;
profile: unknown | null;
loadCandidateRange?: () => Promise<Readonly<{ startTime: string; endTime: string }> | null>;
checkSessionOwned: (sessionId: string) => Promise<boolean>;
checkChartProfileOwned: (chartProfileId: string) => Promise<boolean>;
featureEnabled: boolean;
@@ -211,6 +212,43 @@ export type ReportCreateCoreDeps = Readonly<{
now?: () => Date;
}>;
const candidateClockPattern = /^((?:[01]\d|2[0-3]):[0-5]\d)(?::00(?:\.0+)?)?$/;
function candidateWindow(
range: Readonly<{ startTime: string; endTime: string }> | null,
): { startTime: string; endTime: string } | null {
if (!range) return null;
const startTime = text(range.startTime)?.match(candidateClockPattern)?.[1];
const endTime = text(range.endTime)?.match(candidateClockPattern)?.[1];
if (!startTime || !endTime || startTime > endTime) throw new Error("report_candidate_range_invalid");
return { startTime, endTime };
}
export function resolveReportBirthTimeSensitivityInput(
profileValue: unknown,
birthTimeStatus: ReportSubjectBirthTimeStatus,
representativeTime: string,
candidateRange: Readonly<{ startTime: string; endTime: string }> | null,
): Partial<ConsultationInput> {
void profileValue;
if (birthTimeStatus === "confirmed") {
return { birth_time_accuracy: "confirmed", representative_time: representativeTime };
}
const trustedRange = candidateWindow(candidateRange);
if (trustedRange && trustedRange.startTime === trustedRange.endTime) {
return { birth_time_accuracy: "confirmed", representative_time: representativeTime };
}
const base = {
birth_time_accuracy: birthTimeStatus === "reported" ? "approximate" : "provisional",
representative_time: representativeTime,
} as const;
return trustedRange
? { ...base, candidate_range: { start_time: trustedRange.startTime, end_time: trustedRange.endTime } }
: base;
}
export async function resolveReportCreate(deps: ReportCreateCoreDeps): Promise<ReportRouteResponse> {
// Same-origin gate first (CSRF), then auth.
const originDecision = checkSameOrigin(deps.requestUrl, deps.origin, deps.allowedOrigins, deps.requestHeaders);
@@ -406,6 +444,24 @@ export async function resolveReportCreate(deps: ReportCreateCoreDeps): Promise<R
const finishGeneration = async (): Promise<ReportRouteResponse> => {
// Real workflow evidence (main chain), never mock/example/random data.
let sensitivityInput: Partial<ConsultationInput>;
try {
sensitivityInput = resolveReportBirthTimeSensitivityInput(
profile,
birthTimeStatus,
`${String(birthClock.hour).padStart(2, "0")}:${String(birthClock.minute).padStart(2, "0")}`,
birthTimeStatus === "confirmed" || !deps.loadCandidateRange
? null
: await deps.loadCandidateRange(),
);
} catch {
await deps.persistence.markFailed(userId, row.id, REPORT_STABLE_CODES.calculationUnavailable);
await releaseBilling(REPORT_STABLE_CODES.calculationUnavailable);
return {
status: 502,
body: { error: "出生时间可信区间不可用", code: REPORT_STABLE_CODES.calculationUnavailable },
};
}
const workflowInputs: ConsultationInput[] = payload.themes.map((theme) => ({
year: birthDate.year,
month: birthDate.month,
@@ -420,6 +476,7 @@ export async function resolveReportCreate(deps: ReportCreateCoreDeps): Promise<R
question: `请为个人报告计算 ${theme} 主题证据`,
theme,
entryMode: "direct_chart",
...sensitivityInput,
}));
const workflows: { theme: string; workflow: unknown }[] = [];
+59 -3
View File
@@ -17,7 +17,10 @@ import {
} from "@/lib/personal-report-worker-core";
import { createSupabasePersonalReportService } from "@/lib/personal-report-service";
import { createPersonalReportSectionService } from "@/lib/personal-report-section-service-core";
import { resolveReportBirthClock } from "@/lib/personal-report-route-core";
import {
resolveReportBirthClock,
resolveReportBirthTimeSensitivityInput,
} from "@/lib/personal-report-route-core";
import { createAdminSupabaseClient } from "@/lib/supabase/admin";
import { completeUsage, releaseUsage } from "@/lib/consultation-billing";
import type { ConsultationInput } from "@/mastra/consultation-workflow";
@@ -30,6 +33,11 @@ const PROFILE_COLUMNS = [
"active_birth_time",
"birth_time_source",
"birth_time_status",
"rectification_case_id",
"declared_window_start",
"declared_window_end",
"uncertainty_before_minutes",
"uncertainty_after_minutes",
"latitude",
"longitude",
"timezone_offset",
@@ -88,7 +96,10 @@ function skillSnapshotForReport(context: PersonalReportWorkerGenerationContext):
};
}
async function generateProductionReport(context: PersonalReportWorkerGenerationContext) {
async function generateProductionReport(
context: PersonalReportWorkerGenerationContext,
candidateRange: Readonly<{ startTime: string; endTime: string }> | null = null,
) {
const profile = record(context.profile);
if (!profile) throw new PersonalReportWorkerError("profile_incomplete", false);
@@ -104,6 +115,12 @@ async function generateProductionReport(context: PersonalReportWorkerGenerationC
}
const birthClock = usableBirth.clock;
const birthTimeStatus = usableBirth.status;
const sensitivityInput = resolveReportBirthTimeSensitivityInput(
profile,
birthTimeStatus,
`${String(birthClock.hour).padStart(2, "0")}:${String(birthClock.minute).padStart(2, "0")}`,
candidateRange,
);
const catalog = await loadLanguageModelCatalog();
const model = catalog.models.find((entry) => entry.id === catalog.defaultModelId) ?? null;
@@ -125,6 +142,7 @@ async function generateProductionReport(context: PersonalReportWorkerGenerationC
question: `请为个人报告计算 ${rawTheme} 主题证据`,
theme: rawTheme as ConsultationInput["theme"],
entryMode: "direct_chart",
...sensitivityInput,
};
try {
workflows.push({
@@ -233,7 +251,45 @@ function createProductionWorker(workerId: string) {
if (error) throw new Error("personal report profile load failed");
return data ?? null;
},
generate: generateProductionReport,
generate: async (context) => {
const profile = record(context.profile) ?? {};
if (resolveReportBirthClock(profile)?.status === "confirmed") {
return generateProductionReport(context);
}
const caseId = text(profile.rectification_case_id);
if (caseId) {
const { data, error } = await admin
.from("birth_time_rectification_cases")
.select("candidate_start,candidate_end")
.eq("id", caseId)
.eq("user_id", context.report.userId)
.in("status", ["confirmed", "completed"])
.maybeSingle();
if (error) throw new PersonalReportWorkerError("calculation_unavailable", true);
const row = record(data);
if (typeof row?.candidate_start === "string" && typeof row.candidate_end === "string") {
return generateProductionReport(context, {
startTime: row.candidate_start,
endTime: row.candidate_end,
});
}
}
const { data, error } = await admin
.from("agentic_rectification_cases")
.select("candidate_range,updated_at")
.eq("user_id", context.report.userId)
.eq("status", "candidate_accepted")
.order("updated_at", { ascending: false })
.limit(1)
.maybeSingle();
if (error) throw new PersonalReportWorkerError("calculation_unavailable", true);
const range = record(record(data)?.candidate_range);
return generateProductionReport(context, range
&& typeof range.start_time === "string"
&& typeof range.end_time === "string"
? { startTime: range.start_time, endTime: range.end_time }
: null);
},
});
}
@@ -172,6 +172,25 @@ export type ReportPratyantarTimeline = Readonly<{
current: ReportDashaPeriod | null;
next: ReportDashaPeriod | null;
}>;
export type ReportBirthTimeSensitivityFact = Readonly<{
window: Readonly<{
startTime: string;
endTime: string;
representativeTime: string;
candidateCount: number;
}>;
themes: readonly Readonly<{
theme: string;
status: "stable" | "sensitive";
stableLayers: readonly string[];
sensitiveLayers: readonly string[];
minuteVariations: readonly Readonly<{
layer: string;
values: readonly Readonly<{ minute: string; value: string }>[];
}>[];
}>[];
claimBoundary: string;
}>;
/**
* Server-computed interpretive facts. Every entry is a closed vocabulary value
@@ -188,6 +207,7 @@ export type ReportInterpretiveFacts = Readonly<{
convergenceDomains: readonly string[];
planetaryFriendship: readonly ReportPlanetaryFriendshipFact[];
pratyantarTimeline: ReportPratyantarTimeline | null;
birthTimeSensitivity?: ReportBirthTimeSensitivityFact;
}>;
/**
@@ -376,6 +396,28 @@ const interpretiveFactsSchema = z.object({
current: dashaPeriodSchema.nullable(),
next: dashaPeriodSchema.nullable(),
}).strict().nullable(),
birthTimeSensitivity: z.object({
window: z.object({
startTime: z.string().regex(/^\d{2}:\d{2}$/),
endTime: z.string().regex(/^\d{2}:\d{2}$/),
representativeTime: z.string().regex(/^\d{2}:\d{2}$/),
candidateCount: z.number().int().min(2).max(15),
}).strict(),
themes: z.array(z.object({
theme: idSchema,
status: z.enum(["stable", "sensitive"]),
stableLayers: z.array(seedTextSchema(100)).max(24),
sensitiveLayers: z.array(seedTextSchema(100)).max(24),
minuteVariations: z.array(z.object({
layer: seedTextSchema(100),
values: z.array(z.object({
minute: z.string().regex(/^(?:[01]\d|2[0-3]):[0-5]\d$/),
value: seedTextSchema(300),
}).strict()).min(2).max(15),
}).strict()).max(24),
}).strict()).max(12),
claimBoundary: seedTextSchema(500),
}).strict().optional(),
}).strict();
const themeNarrativeSeedSchema = z.object({
@@ -567,6 +609,24 @@ function sortedBundleContent(bundle: Omit<ReportEvidenceBundleV2, "bundleHash">)
}))
.sort((a, b) => a.planet.localeCompare(b.planet)),
pratyantarTimeline: bundle.interpretiveFacts.pratyantarTimeline,
...(bundle.interpretiveFacts.birthTimeSensitivity ? {
birthTimeSensitivity: {
...bundle.interpretiveFacts.birthTimeSensitivity,
themes: [...bundle.interpretiveFacts.birthTimeSensitivity.themes]
.map((theme) => ({
...theme,
stableLayers: sortedUnique(theme.stableLayers),
sensitiveLayers: sortedUnique(theme.sensitiveLayers),
minuteVariations: [...theme.minuteVariations]
.map((variation) => ({
...variation,
values: [...variation.values].sort((a, b) => a.minute.localeCompare(b.minute)),
}))
.sort((a, b) => a.layer.localeCompare(b.layer)),
}))
.sort((a, b) => a.theme.localeCompare(b.theme)),
},
} : {}),
},
themeNarrativeSeeds: [...bundle.themeNarrativeSeeds]
.map((seed) => ({ ...seed, evidenceRefs: sortedUnique(seed.evidenceRefs) }))
@@ -19,6 +19,13 @@ export const consultationInputSchema = z.object({
entryMode: z.enum(["direct_chart", "rectification"]).default("direct_chart"),
declared_accuracy: z.enum(["rectified", "minute", "15min", "1hour", "unknown"]).optional(),
time_source: z.string().trim().min(1).max(40).optional(),
birth_time_accuracy: z.enum(["confirmed", "provisional", "approximate"]).optional(),
candidate_range: z.object({ start_time: z.string(), end_time: z.string() }).passthrough().optional(),
representative_time: z.string().optional(),
declared_window_start: z.string().optional(),
declared_window_end: z.string().optional(),
uncertainty_before_minutes: z.number().int().min(0).max(720).optional(),
uncertainty_after_minutes: z.number().int().min(0).max(720).optional(),
});
export type ConsultationInput = z.infer<typeof consultationInputSchema>;
type JsonRecord = Record<string, unknown>;
@@ -41,6 +48,10 @@ export const consultationWorkflowResponseSchema = z.object({
chart: z.record(z.unknown()),
routing: z.record(z.unknown()),
consumer_context: workflowConsumerContextSchema,
birth_time_sensitivity: z.object({
schema: z.literal("jyotish.report_birth_time_sensitivity.v1"),
status: z.enum(["not_applicable", "candidate_window_only"]),
}).passthrough().optional(),
}).passthrough();
function record(value: unknown): JsonRecord {
@@ -469,6 +469,72 @@ test("core create: 201 for a reported minute uses that clock and directional pol
assert.equal(bundle.answerPolicy.birthTimePolicy, "reported_directional_only");
});
test("core create: confirmed reports do not query rectification ranges", async () => {
const response = await resolveReportCreate(baseDeps({
loadCandidateRange: async () => { throw new Error("must not query"); },
}));
assert.equal(response.status, 201);
});
test("core create: trusted rectification range wins and accepted multi-minute input stays provisional", async () => {
const inputs: Record<string, unknown>[] = [];
const response = await resolveReportCreate(baseDeps({
profile: profileFixture({
birth_time_status: "accepted",
declared_window_start: "09:00",
declared_window_end: "11:00",
uncertainty_before_minutes: 30,
uncertainty_after_minutes: 30,
}),
loadCandidateRange: async () => ({ startTime: "10:00:00", endTime: "10:04:00" }),
runWorkflow: async (input) => {
inputs.push(input as unknown as Record<string, unknown>);
return chartPayload();
},
}));
assert.equal(response.status, 201);
assert.equal(inputs[0].birth_time_accuracy, "provisional");
assert.deepEqual(inputs[0].candidate_range, { start_time: "10:00", end_time: "10:04" });
assert.equal("declared_window_start" in inputs[0], false);
assert.equal("uncertainty_before_minutes" in inputs[0], false);
});
test("core create: accepted single-minute range becomes confirmed without a candidate window", async () => {
const inputs: Record<string, unknown>[] = [];
const response = await resolveReportCreate(baseDeps({
profile: profileFixture({ birth_time_status: "accepted" }),
loadCandidateRange: async () => ({ startTime: "10:00", endTime: "10:00" }),
runWorkflow: async (input) => {
inputs.push(input as unknown as Record<string, unknown>);
return chartPayload();
},
}));
assert.equal(response.status, 201);
assert.equal(inputs[0].birth_time_accuracy, "confirmed");
assert.equal("candidate_range" in inputs[0], false);
});
test("core create: no adopted range uses the fixed accuracy fallback", async () => {
const inputs: Record<string, unknown>[] = [];
const response = await resolveReportCreate(baseDeps({
profile: profileFixture({
birth_time_status: "accepted",
declared_window_start: "10:00",
declared_window_end: "10:00",
uncertainty_before_minutes: 0,
uncertainty_after_minutes: 0,
}),
runWorkflow: async (input) => {
inputs.push(input as unknown as Record<string, unknown>);
return chartPayload();
},
}));
assert.equal(response.status, 201);
assert.equal(inputs[0].birth_time_accuracy, "provisional");
assert.equal("declared_window_start" in inputs[0], false);
assert.equal("uncertainty_before_minutes" in inputs[0], false);
});
test("core create: 422 birth_time_not_usable for incomplete profile fields", async () => {
const response = await resolveReportCreate(baseDeps({
profile: profileFixture({ latitude: null, longitude: null }),
@@ -1139,6 +1205,13 @@ test("POST route uses dual clients: authenticated reads + admin persistence", ()
assert.doesNotMatch(createRoute, /not wired yet|尚未就绪/);
});
test("POST route only trusts terminal legacy rectification ranges", () => {
assert.match(
createRoute,
/from\("birth_time_rectification_cases"\)[\s\S]*?\.in\("status", \["confirmed", "completed"\]\)/,
);
});
test("GET/DELETE use the authenticated client (least privilege) and the core handlers", () => {
assert.match(itemRoute, /createServerSupabaseClient\(\)/);
assert.match(itemRoute, /createSupabasePersonalReportService\(supabase\)/);
@@ -175,6 +175,36 @@ test("expandedByDefault=true expands the appendix even in default mode", () => {
assert.match(markup, /<details class="personal-report-appendix-details" open="">/, "expandedByDefault forces the expanded state");
});
test("birth-time sensitivity alone keeps the appendix visible and shows minute changes", () => {
const document = structuredClone(canonicalV2Fixture);
document.evidenceAppendix = {
expandedByDefault: false,
techniqueAudit: [],
conflicts: [],
calculationEvidence: [],
blockedTechniques: [],
birthTimeSensitivity: {
window: { startTime: "10:00", endTime: "10:01", representativeTime: "10:00", candidateCount: 2 },
themes: [{
theme: "career",
status: "sensitive",
stableLayers: [],
sensitiveLayers: ["D10.ascendant"],
minuteVariations: [{
layer: "D10.ascendant",
values: [{ minute: "10:00", value: "Leo" }, { minute: "10:01", value: "Virgo" }],
}],
}],
claimBoundary: "Candidate-window comparison only.",
},
};
const markup = render(document);
assert.match(markup, /personal-report-appendix-details/);
assert.match(markup, /D10\.ascendant/);
assert.match(markup, /10:00=Leo/);
assert.match(markup, /10:01=Virgo/);
});
test("D1 SVG renders when real houses exist; D9/D10 are never fabricated", () => {
const svgCount = (markup: string) => (markup.match(/<svg/g) ?? []).length;
@@ -671,5 +671,9 @@ test("instrumentation retains the Skill guard and starts a singleton Node worker
);
assert.match(productionAdapter, /jyotishaPersonalReportWorker/);
assert.match(productionAdapter, /if \(state\.jyotishaPersonalReportWorker\) return/);
assert.match(
productionAdapter,
/from\("birth_time_rectification_cases"\)[\s\S]*?\.in\("status", \["confirmed", "completed"\]\)/,
);
assert.doesNotMatch(productionAdapter, /after\s*\(/);
});
@@ -123,6 +123,7 @@ type FixtureOptions = Readonly<{
pollutedNames?: boolean;
structuredVargas?: boolean;
timingReady?: boolean;
birthTimeSensitivity?: boolean;
}>;
function workflowFixture(options: FixtureOptions = {}): JsonRecord {
@@ -289,6 +290,30 @@ function workflowFixture(options: FixtureOptions = {}): JsonRecord {
answer_policy: { can_answer_direction: true, can_answer_precise_timing: false },
},
machine_evidence_packet: { sections: machineSections(), conflicts: [] },
...(options.birthTimeSensitivity ? {
birth_time_sensitivity: {
schema: "jyotish.report_birth_time_sensitivity.v1",
status: "candidate_window_only",
window: {
start_time: "10:00",
end_time: "10:01",
representative_time: "10:00",
candidate_count: 2,
},
theme_sensitivity: {
career: { status: "sensitive", stable_layers: ["arudha.A10"], sensitive_layers: ["D10.ascendant"] },
marriage: { status: "sensitive", stable_layers: ["arudha.UL"], sensitive_layers: ["D9.ascendant"] },
},
sensitive_evidence: {
evidence_keys: ["D10.ascendant", "D9.ascendant"],
layers: {
"D10.ascendant": { "10:01": "Virgo", "10:00": "Leo" },
"D9.ascendant": { "10:01": "Taurus", "10:00": "Aries" },
},
},
claim_boundary: "Candidate-window comparison only; no minute is selected or confirmed.",
},
} : {}),
};
}
@@ -420,6 +445,60 @@ test("bundle hash is stable under interpretive field ordering", () => {
assert.equal(shuffled.bundleHash, bundle.bundleHash);
});
test("birth-time sensitivity carries actual minute changes and hashes independently of ordering", () => {
const workflow = workflowFixture({ birthTimeSensitivity: true });
const build = (value: JsonRecord) => buildReportEvidenceBundleV2({
workflows: ["career", "marriage"].map((theme) => ({ theme, workflow: value })),
subject: { displayName: "冒烟用户", birthTimeStatus: "accepted", birthPlaceLabel: "冒烟市" },
requestedThemes: ["career", "marriage"],
reportType: "personal_full",
presentationMode: "default",
skillSnapshot: { name: "jyotish-personal-report", version: "1.0.0", sha256: "a".repeat(64), sourceCommit: null },
});
const bundle = build(workflow);
const sensitivity = bundle.interpretiveFacts.birthTimeSensitivity;
assert.ok(sensitivity);
assert.deepEqual(sensitivity.themes.find((theme) => theme.theme === "career")?.minuteVariations, [{
layer: "D10.ascendant",
values: [{ minute: "10:00", value: "Leo" }, { minute: "10:01", value: "Virgo" }],
}]);
const shuffledWorkflow = structuredClone(workflow);
const packet = shuffledWorkflow.birth_time_sensitivity as JsonRecord;
packet.theme_sensitivity = Object.fromEntries(Object.entries(packet.theme_sensitivity as JsonRecord).reverse());
const layers = (packet.sensitive_evidence as JsonRecord).layers as JsonRecord;
layers["D10.ascendant"] = Object.fromEntries(Object.entries(layers["D10.ascendant"] as JsonRecord).reverse());
assert.equal(build(shuffledWorkflow).bundleHash, bundle.bundleHash);
const plan = buildPersonalReportSectionPlan(bundle, "standard");
const careerSection = plan.sections.find((section) => section.theme === "career")!;
const careerBundle = filterReportEvidenceBundleForSection(bundle, careerSection);
assert.deepEqual(careerBundle.interpretiveFacts.birthTimeSensitivity?.themes.map((theme) => theme.theme), ["career"]);
assert.deepEqual(
careerBundle.interpretiveFacts.birthTimeSensitivity?.themes.flatMap((theme) => theme.minuteVariations.map((row) => row.layer)),
["D10.ascendant"],
);
});
test("confirmed not-applicable sensitivity leaves the legacy bundle hash unchanged", () => {
const baseline = buildBundle();
const workflow = workflowFixture();
workflow.birth_time_sensitivity = {
schema: "jyotish.report_birth_time_sensitivity.v1",
status: "not_applicable",
accuracy: "confirmed",
};
const confirmed = buildReportEvidenceBundleV2({
workflows: THEMES.map((theme) => ({ theme, workflow })),
subject: { displayName: "冒烟用户", birthTimeStatus: "reported", birthPlaceLabel: "冒烟市" },
requestedThemes: [...THEMES],
reportType: "personal_full",
presentationMode: "default",
skillSnapshot: { name: "jyotish-personal-report", version: "1.0.0", sha256: "a".repeat(64), sourceCommit: null },
});
assert.equal(confirmed.bundleHash, baseline.bundleHash);
});
// ---------------------------------------------------------------------------
// Task 1: fail-closed validation of the new fields
// ---------------------------------------------------------------------------
@@ -0,0 +1,53 @@
"""Build the report-facing projection of birth-time sensitivity evidence."""
from __future__ import annotations
from copy import deepcopy
from hashlib import sha256
import json
from typing import Any, Mapping
try:
from flexible_birth_time_profile import _candidate_window_authority_violation
except ImportError: # pragma: no cover - package import
from scripts.flexible_birth_time_profile import _candidate_window_authority_violation
SCHEMA_VERSION = "jyotish.flexible_birth_time_full_report_projection.v1"
SUPPORT_SCHEMA_VERSION = "jyotish.flexible_birth_time_report_support.v1"
class FlexibleBirthTimeFullReportProjectionError(ValueError):
"""Raised when report support cannot be projected safely."""
def build_flexible_birth_time_full_report_projection(support: Mapping[str, Any]) -> dict[str, Any]:
packet = _validate_support(support)
window = deepcopy(dict(packet.get("birth_time_window") or {}))
return {
"schema_version": SCHEMA_VERSION,
"projection_id": _digest_id(str(packet.get("support_id")), json.dumps(window, sort_keys=True, separators=(",", ":"))),
"report_section_type": "birth_time_sensitivity",
"window": window,
"stable_structure_section": deepcopy(dict(packet.get("stable_report_evidence") or {})),
"minute_sensitive_section": deepcopy(dict(packet.get("sensitive_report_evidence") or {})),
"theme_sensitivity": deepcopy(dict(packet.get("theme_sensitivity") or {})),
"trace": deepcopy(list(packet.get("trace") or [])),
"status": "candidate_window_only",
"claim_boundary": str(packet.get("claim_boundary") or "Sensitivity remains conditional."),
}
def _validate_support(value: Mapping[str, Any]) -> dict[str, Any]:
violation = _candidate_window_authority_violation(value)
if violation:
raise FlexibleBirthTimeFullReportProjectionError(f"candidate_window_authority_forbidden:{violation}")
if not isinstance(value, Mapping) or value.get("schema_version") != SUPPORT_SCHEMA_VERSION:
raise FlexibleBirthTimeFullReportProjectionError("flexible_birth_time_report_support_schema_invalid")
if value.get("status") != "candidate_window_only":
raise FlexibleBirthTimeFullReportProjectionError("support_must_remain_candidate_window_only")
return dict(value)
def _digest_id(*parts: str) -> str:
payload = json.dumps(parts, ensure_ascii=True, separators=(",", ":"))
return f"flex-report-projection://{sha256(payload.encode('utf-8')).hexdigest()[:24]}"
+323
View File
@@ -0,0 +1,323 @@
"""Build a read-only sensitivity profile for an unresolved birth-time window."""
from __future__ import annotations
from copy import deepcopy
from datetime import datetime
from hashlib import sha256
import json
import re
from typing import Any, Mapping, Sequence
SCHEMA_VERSION = "jyotish.flexible_birth_time_profile.v1"
MAX_CANDIDATE_MINUTES = 15
_CLOCK_RE = re.compile(r"(?:[01]\d|2[0-3]):[0-5]\d")
_PROHIBITED_AUTHORITY_FIELDS = frozenset({
"approved_birth_time",
"final_birth_time",
"winner",
"approval",
"approval_authority",
})
_PROHIBITED_SOURCE_STATUSES = frozenset({"approved", "confirmed"})
class FlexibleBirthTimeProfileError(ValueError):
"""Raised when a candidate window cannot be represented safely."""
def build_flexible_birth_time_profile(
candidates: Sequence[Mapping[str, Any]],
*,
source_reference: Mapping[str, Any],
) -> dict[str, Any]:
_reject_candidate_window_authority(candidates)
rows = _normalize_candidates(candidates)
source = _normalize_source_reference(source_reference)
stable: dict[str, Any] = {}
sensitive: dict[str, dict[str, Any]] = {}
for key in sorted({key for row in rows for key in row["evidence"]}):
values = {row["candidate_time"]: row["evidence"].get(key) for row in rows}
if len({_canonical(value) for value in values.values()}) == 1:
stable[key] = deepcopy(next(iter(values.values())))
else:
sensitive[key] = deepcopy(values)
times = [row["candidate_time"] for row in rows]
profile_id = _profile_id(times, source["review_id"])
return {
"schema_version": SCHEMA_VERSION,
"flexible_profile_id": profile_id,
"birth_time_window": {
"start_time": times[0],
"end_time": times[-1],
"candidate_count": len(times),
"candidate_times": times,
},
"candidate_references": [
{"candidate_id": row["candidate_id"], "candidate_time": row["candidate_time"]}
for row in rows
],
"stable_evidence": stable,
"sensitive_evidence": sensitive,
"source_reference": source,
"trace": [
{"kind": "flexible_birth_time_profile", "reference": profile_id},
{"kind": "candidate_window", "reference": source["review_id"]},
*(
{"kind": "candidate", "reference": f"candidate://{row['candidate_id']}"}
for row in rows
),
],
"status": "candidate_window_only",
"claim_boundary": (
"Read-only candidate-window comparison. It cannot select or confirm a birth minute, "
"replace chart identity, or grant authority to a candidate chart."
),
}
def build_flexible_birth_time_profile_from_window(
*,
birth_date: str,
start_time: str,
end_time: str,
candidate_times: Sequence[str],
lat: float,
lon: float,
tz: float,
ayanamsa: str = "raman",
node_mode: str = "mean",
source_reference: Mapping[str, Any],
) -> dict[str, Any]:
start, end = _parse_window(birth_date, start_time, end_time)
normalized_times = _normalize_candidate_times(birth_date, start, end, candidate_times)
candidates = [
_candidate_from_recast(
candidate_at=value,
recast=_recast_candidate_layers(
value, lat=lat, lon=lon, tz=tz, ayanamsa=ayanamsa, node_mode=node_mode,
),
ayanamsa=ayanamsa,
node_mode=node_mode,
)
for value in normalized_times
]
profile = build_flexible_birth_time_profile(candidates, source_reference=source_reference)
profile["calculation_profile"] = {
"ayanamsa": ayanamsa,
"node_mode": node_mode,
"coordinate_mode": "explicit_lat_lon_tz",
"candidate_recast": "native_domain_calculation_service",
}
return profile
def _recast_candidate_layers(
candidate: datetime,
*,
lat: float,
lon: float,
tz: float,
ayanamsa: str,
node_mode: str,
) -> dict[str, Any]:
try:
import domain_calculation_service
import jaimini
import kp_system
import varga
except ModuleNotFoundError: # pragma: no cover - package import
from scripts import domain_calculation_service, jaimini, kp_system, varga
chart = domain_calculation_service.compute_chart({
"year": candidate.year,
"month": candidate.month,
"day": candidate.day,
"hour": candidate.hour,
"minute": candidate.minute,
"second": candidate.second,
"lat": lat,
"lon": lon,
"tz": tz,
"ayanamsa": ayanamsa,
"node_mode": node_mode,
})
planets = {
name: row["lon"]
for name, row in (chart.get("planets") or {}).items()
if name in {"Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu"}
}
ascendant = chart.get("ascendant") or {}
asc_lon = float(ascendant["lon"])
divisions = [2, 3, 4, 6, 7, 8, 9, 10, 11, 12, 16, 24, 30, 40, 45, 60]
vargas = varga.calc_all_vargas(planets, asc_lon, divisions=divisions)
arudha = jaimini.calc_arudha_padas(int(asc_lon // 30), planets)
padas = arudha.get("padas") or {}
kp_cusps: dict[str, Any] = {}
for house_key in ("house_1", "house_4", "house_7", "house_10"):
degree = ((chart.get("houses") or {}).get(house_key) or {}).get("cusp_degree")
if degree is None:
continue
lords = kp_system.get_kp_lords(float(degree))
kp_cusps[house_key] = {
"sign": lords.get("sign"),
"nakshatra_lord": lords.get("nakshatra_lord"),
"sub_lord": lords.get("sub_lord"),
}
return {
"ascendant": ascendant,
"varga_lagna": {key: value.get("Ascendant") or {} for key, value in vargas.items()},
"arudha": {"A7": padas.get("A7") or {}, "A10": padas.get("A10") or {}, "UL": arudha.get("upapada") or {}},
"kp_cusps": kp_cusps,
}
def _candidate_from_recast(
*,
candidate_at: datetime,
recast: Mapping[str, Any],
ayanamsa: str,
node_mode: str,
) -> dict[str, Any]:
evidence: dict[str, Any] = {"D1.ascendant": (recast.get("ascendant") or {}).get("sign")}
for key, value in (recast.get("varga_lagna") or {}).items():
if isinstance(key, str) and key.startswith("D") and isinstance(value, Mapping):
evidence[f"{key.split('_', 1)[0]}.ascendant"] = value.get("sign")
for key in ("A7", "A10", "UL"):
value = (recast.get("arudha") or {}).get(key)
if isinstance(value, Mapping):
evidence[f"arudha.{key}"] = value.get("sign")
evidence["KP.cusp_observation"] = recast.get("kp_cusps") or {}
candidate_time = candidate_at.strftime("%H:%M")
candidate_id = f"candidate-{candidate_time.replace(':', '')}"
return {
"candidate_id": candidate_id,
"candidate_time": candidate_time,
"evidence": evidence,
"trace": [
{"kind": "candidate_chart_recast", "reference": f"candidate-chart://{candidate_id}"},
{"kind": "calculation_profile", "reference": f"ayanamsa://{ayanamsa}/node/{node_mode}"},
],
}
def _parse_window(birth_date: str, start_time: str, end_time: str) -> tuple[datetime, datetime]:
if not _is_hh_mm(start_time) or not _is_hh_mm(end_time):
raise FlexibleBirthTimeProfileError("birth_date_or_candidate_time_invalid")
try:
start = datetime.strptime(f"{birth_date} {start_time}", "%Y-%m-%d %H:%M")
end = datetime.strptime(f"{birth_date} {end_time}", "%Y-%m-%d %H:%M")
except (TypeError, ValueError) as exc:
raise FlexibleBirthTimeProfileError("birth_date_or_candidate_time_invalid") from exc
if end < start:
raise FlexibleBirthTimeProfileError("candidate_window_must_not_cross_midnight")
return start, end
def _normalize_candidate_times(
birth_date: str,
start: datetime,
end: datetime,
candidate_times: Sequence[str],
) -> list[datetime]:
if isinstance(candidate_times, (str, bytes)) or not isinstance(candidate_times, Sequence):
raise FlexibleBirthTimeProfileError("candidate_times_required")
if len(candidate_times) < 2 or len(candidate_times) > MAX_CANDIDATE_MINUTES:
raise FlexibleBirthTimeProfileError("candidate_count_must_be_two_to_fifteen")
values: list[datetime] = []
for raw in candidate_times:
if not _is_hh_mm(raw):
raise FlexibleBirthTimeProfileError("candidate_time_invalid")
try:
value = datetime.strptime(f"{birth_date} {raw}", "%Y-%m-%d %H:%M")
except (TypeError, ValueError) as exc:
raise FlexibleBirthTimeProfileError("candidate_time_invalid") from exc
if value < start or value > end:
raise FlexibleBirthTimeProfileError("candidate_time_outside_window")
values.append(value)
if len(set(values)) != len(values):
raise FlexibleBirthTimeProfileError("candidate_times_must_be_unique")
return sorted(values)
def _normalize_candidates(candidates: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
if isinstance(candidates, (str, bytes)) or not isinstance(candidates, Sequence):
raise FlexibleBirthTimeProfileError("candidates_required")
if len(candidates) < 2 or len(candidates) > MAX_CANDIDATE_MINUTES:
raise FlexibleBirthTimeProfileError("candidate_count_must_be_two_to_fifteen")
rows: list[dict[str, Any]] = []
for candidate in candidates:
if not isinstance(candidate, Mapping):
raise FlexibleBirthTimeProfileError("candidate_must_be_mapping")
candidate_id = candidate.get("candidate_id")
candidate_time = candidate.get("candidate_time")
evidence = candidate.get("evidence")
trace = candidate.get("trace")
if not isinstance(candidate_id, str) or not candidate_id:
raise FlexibleBirthTimeProfileError("candidate_id_required")
if not _is_hh_mm(candidate_time):
raise FlexibleBirthTimeProfileError("candidate_time_required")
if not isinstance(evidence, Mapping) or not evidence:
raise FlexibleBirthTimeProfileError("candidate_evidence_required")
if not isinstance(trace, list) or not trace:
raise FlexibleBirthTimeProfileError("candidate_trace_required")
rows.append({"candidate_id": candidate_id, "candidate_time": candidate_time, "evidence": dict(evidence), "trace": trace})
if len({row["candidate_id"] for row in rows}) != len(rows):
raise FlexibleBirthTimeProfileError("candidate_ids_must_be_unique")
if len({row["candidate_time"] for row in rows}) != len(rows):
raise FlexibleBirthTimeProfileError("candidate_times_must_be_unique")
return sorted(rows, key=lambda row: row["candidate_time"])
def _normalize_source_reference(value: Mapping[str, Any]) -> dict[str, str]:
if not isinstance(value, Mapping):
raise FlexibleBirthTimeProfileError("source_reference_must_be_mapping")
_reject_candidate_window_authority(value)
status = value.get("status")
if isinstance(status, str) and status.strip().lower() in _PROHIBITED_SOURCE_STATUSES:
raise FlexibleBirthTimeProfileError("approved_or_confirmed_source_reference_forbidden")
review_id = value.get("review_id")
if not isinstance(review_id, str) or not review_id:
raise FlexibleBirthTimeProfileError("source_review_id_required")
return {"review_id": review_id, "status": "review_required"}
def _profile_id(candidate_times: Sequence[str], review_id: str) -> str:
digest = sha256(repr((tuple(candidate_times), review_id)).encode("utf-8")).hexdigest()[:24]
return f"flexible-birth-time://{digest}"
def _canonical(value: Any) -> str:
return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), default=str)
def _is_hh_mm(value: Any) -> bool:
return isinstance(value, str) and _CLOCK_RE.fullmatch(value) is not None
def _candidate_window_authority_violation(value: Any, path: str = "$") -> str | None:
if isinstance(value, Mapping):
for raw_key, item in value.items():
key = str(raw_key).strip().lower()
child_path = f"{path}.{raw_key}"
if key in _PROHIBITED_AUTHORITY_FIELDS:
return child_path
if key == "source_reference" and isinstance(item, Mapping):
status = item.get("status")
if isinstance(status, str) and status.strip().lower() in _PROHIBITED_SOURCE_STATUSES:
return f"{child_path}.status"
violation = _candidate_window_authority_violation(item, child_path)
if violation:
return violation
elif isinstance(value, (list, tuple)):
for index, item in enumerate(value):
violation = _candidate_window_authority_violation(item, f"{path}[{index}]")
if violation:
return violation
return None
def _reject_candidate_window_authority(value: Any) -> None:
violation = _candidate_window_authority_violation(value)
if violation:
raise FlexibleBirthTimeProfileError(f"candidate_window_authority_forbidden:{violation}")
@@ -0,0 +1,63 @@
"""Render a bounded birth-time sensitivity appendix section."""
from __future__ import annotations
from typing import Any, Mapping
try:
from flexible_birth_time_profile import _candidate_window_authority_violation
except ImportError: # pragma: no cover - package import
from scripts.flexible_birth_time_profile import _candidate_window_authority_violation
SCHEMA_VERSION = "jyotish.flexible_birth_time_full_report_projection.v1"
class FlexibleBirthTimeReportSectionError(ValueError):
"""Raised when a sensitivity projection cannot be rendered safely."""
def render_flexible_birth_time_report_section(projection: Mapping[str, Any]) -> str:
packet = _validate_projection(projection)
window = packet.get("window") or {}
stable = packet.get("stable_structure_section") or {}
sensitive = packet.get("minute_sensitive_section") or {}
lines = [
"### 出生时间敏感度",
"",
f"- 可信区间:{window.get('start_time')}{window.get('end_time')}",
f"- 代表分钟:{window.get('representative_time')}",
f"- 候选分钟数:{window.get('candidate_count')}",
"",
"#### 窗口内稳定层",
]
for key in stable.get("evidence_keys") or []:
lines.append(f"- `{key}`{_render_value((stable.get('layers') or {}).get(key))}")
lines.extend(["", "#### 窗口内敏感层"])
for key in sensitive.get("evidence_keys") or []:
lines.append(f"- `{key}`{_render_value((sensitive.get('layers') or {}).get(key))}")
lines.extend([
"",
"#### 使用边界",
"",
"- 被标记为敏感的主题只能作条件性解读,不能据此认定唯一出生分钟。",
])
return "\n".join(lines).rstrip() + "\n"
def _validate_projection(value: Mapping[str, Any]) -> dict[str, Any]:
violation = _candidate_window_authority_violation(value)
if violation:
raise FlexibleBirthTimeReportSectionError(f"candidate_window_authority_forbidden:{violation}")
if not isinstance(value, Mapping) or value.get("schema_version") != SCHEMA_VERSION:
raise FlexibleBirthTimeReportSectionError("flexible_birth_time_full_report_projection_schema_invalid")
if value.get("status") != "candidate_window_only":
raise FlexibleBirthTimeReportSectionError("projection_must_remain_candidate_window_only")
return dict(value)
def _render_value(value: Any) -> str:
if isinstance(value, Mapping):
return " / ".join(f"{key}={_render_value(item)}" for key, item in value.items())
if isinstance(value, list):
return " / ".join(_render_value(item) for item in value)
return str(value)
@@ -0,0 +1,88 @@
"""Project candidate-window evidence into bounded report support."""
from __future__ import annotations
from copy import deepcopy
from hashlib import sha256
import json
from typing import Any, Mapping
try:
from flexible_birth_time_profile import _candidate_window_authority_violation
except ImportError: # pragma: no cover - package import
from scripts.flexible_birth_time_profile import _candidate_window_authority_violation
SCHEMA_VERSION = "jyotish.flexible_birth_time_report_support.v1"
PROFILE_SCHEMA_VERSION = "jyotish.flexible_birth_time_profile.v1"
THEME_LAYERS: dict[str, tuple[str, ...]] = {
"career": ("D10.ascendant", "arudha.A10"),
"marriage": ("D7.ascendant", "D9.ascendant", "arudha.A7", "arudha.UL"),
"wealth": ("D2.ascendant", "D11.ascendant"),
"health": ("D6.ascendant", "D8.ascendant", "D30.ascendant"),
"timing": ("D1.ascendant", "D60.ascendant", "KP.cusp_observation"),
"general": (),
}
class FlexibleBirthTimeReportSupportError(ValueError):
"""Raised when an unresolved profile cannot form report support."""
def build_flexible_birth_time_report_support(profile: Mapping[str, Any]) -> dict[str, Any]:
packet = _validate_profile(profile)
stable = deepcopy(dict(packet.get("stable_evidence") or {}))
sensitive = deepcopy(dict(packet.get("sensitive_evidence") or {}))
themes = {
theme: _theme_status(theme, stable, sensitive)
for theme in THEME_LAYERS
}
support_id = _digest_id(
str(packet["flexible_profile_id"]),
str((packet.get("birth_time_window") or {}).get("start_time")),
str((packet.get("birth_time_window") or {}).get("end_time")),
)
return {
"schema_version": SCHEMA_VERSION,
"support_id": support_id,
"birth_time_window": deepcopy(dict(packet["birth_time_window"])),
"stable_report_evidence": {"evidence_keys": sorted(stable), "layers": stable},
"sensitive_report_evidence": {"evidence_keys": sorted(sensitive), "layers": sensitive},
"theme_sensitivity": themes,
"trace": deepcopy(list(packet.get("trace") or [])),
"status": "candidate_window_only",
"claim_boundary": (
"Sensitivity support only. Sensitive themes must remain conditional and this packet "
"cannot identify, rank, or confirm a birth minute."
),
}
def _theme_status(theme: str, stable: Mapping[str, Any], sensitive: Mapping[str, Any]) -> dict[str, Any]:
configured = THEME_LAYERS[theme]
keys = tuple(sorted(set(stable) | set(sensitive))) if theme == "general" else configured
sensitive_layers = [key for key in keys if key in sensitive]
stable_layers = [key for key in keys if key in stable]
return {
"status": "sensitive" if sensitive_layers else "stable",
"sensitive_layers": sensitive_layers,
"stable_layers": stable_layers,
}
def _validate_profile(value: Mapping[str, Any]) -> dict[str, Any]:
violation = _candidate_window_authority_violation(value)
if violation:
raise FlexibleBirthTimeReportSupportError(f"candidate_window_authority_forbidden:{violation}")
if not isinstance(value, Mapping) or value.get("schema_version") != PROFILE_SCHEMA_VERSION:
raise FlexibleBirthTimeReportSupportError("flexible_birth_time_profile_schema_invalid")
if value.get("status") != "candidate_window_only":
raise FlexibleBirthTimeReportSupportError("profile_must_remain_candidate_window_only")
if not isinstance(value.get("birth_time_window"), Mapping):
raise FlexibleBirthTimeReportSupportError("birth_time_window_required")
return dict(value)
def _digest_id(*parts: str) -> str:
payload = json.dumps(parts, ensure_ascii=True, separators=(",", ":"))
return f"flex-report-support://{sha256(payload.encode('utf-8')).hexdigest()[:24]}"
+25 -1
View File
@@ -2155,6 +2155,11 @@ def execute_consultation_workflow(
from scripts.reference_transparency_contract import build_reference_transparency_contract
birth_payload = handler._high_rigor_birth_payload(body)
sensitivity_args = type('BirthTimeSensitivityArgs', (), birth_payload)()
try:
birth_time_sensitivity = _load_local_module('jyotish_engine')._build_birth_time_sensitivity(sensitivity_args)
except ValueError as exc:
raise BadRequest(str(exc)) from exc
themes = handler._high_rigor_requested_themes(body)
events = handler._high_rigor_events(body)
question = body.get('question') or ''
@@ -2225,6 +2230,7 @@ def execute_consultation_workflow(
]
if body.get('dry_run') or body.get('plan_only'):
result = handler._high_rigor_workflow_plan_only(birth_payload, themes, events)
result['birth_time_sensitivity'] = birth_time_sensitivity
result['endpoint'] = 'consultation_workflow'
result['entry_mode'] = entry_mode
result['routing'] = route_packet
@@ -2506,6 +2512,7 @@ def execute_consultation_workflow(
'consumer_context': consumer_context,
'western_evidence_packet': western_evidence_packet or {},
'real_case_calibration': real_case_calibration,
'birth_time_sensitivity': birth_time_sensitivity,
'runtime_evidence_log': runtime_evidence_log,
'next_questions': handler._high_rigor_next_questions(rectification, historical_backtest),
'boundary': (
@@ -4745,6 +4752,13 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
'node_mode': body.get('node_mode', body.get('nodeMode', 'mean')),
'today': body.get('today') or body.get('current_date'),
'transit_date': body.get('transit_date') or body.get('reference_date'),
'birth_time_accuracy': body.get('birth_time_accuracy', 'confirmed'),
'candidate_range': body.get('candidate_range'),
'representative_time': body.get('representative_time'),
'declared_window_start': body.get('declared_window_start'),
'declared_window_end': body.get('declared_window_end'),
'uncertainty_before_minutes': body.get('uncertainty_before_minutes'),
'uncertainty_after_minutes': body.get('uncertainty_after_minutes'),
}
def _high_rigor_requested_themes(self, body):
@@ -6556,8 +6570,18 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
'today': body.get('today') or body.get('current_date'),
'transit_date': body.get('transit_date'),
'target_year': body.get('target_year'),
'birth_time_accuracy': body.get('birth_time_accuracy', 'confirmed'),
'candidate_range': body.get('candidate_range'),
'representative_time': body.get('representative_time'),
'declared_window_start': body.get('declared_window_start'),
'declared_window_end': body.get('declared_window_end'),
'uncertainty_before_minutes': body.get('uncertainty_before_minutes'),
'uncertainty_after_minutes': body.get('uncertainty_after_minutes'),
})()
result = engine.cmd_full_reading(args)
try:
result = engine.cmd_full_reading(args)
except ValueError as exc:
raise BadRequest(str(exc)) from exc
if not isinstance(result, dict) or not isinstance(result.get('modules'), dict):
raise BadRequest('full-reading did not return modules')
return result
+130 -198
View File
@@ -2041,7 +2041,6 @@ def _event_replay_contract(args) -> dict:
}
path = getattr(args, 'event_replay_file', None)
if not path:
candidate_segment_table = _load_1993_candidate_segment_table(args)
return {
'schema': 'jyotish.event_replay_contract.v1',
'status': 'blocked',
@@ -2052,12 +2051,6 @@ def _event_replay_contract(args) -> dict:
'Parent/family events may be entered with domain=family and compared against D12 only as '
'candidate discrimination evidence; D12 cannot select a minute by itself.'
),
'candidate_segment_table': candidate_segment_table,
'candidate_segment_reference': (
'candidate_segment_table_not_imported'
if candidate_segment_table
else None
),
'blind_holdout_policy': blind_holdout_policy,
'boundary': 'No user-known events were supplied. This report does not claim retrospective calibration.',
}
@@ -2066,19 +2059,11 @@ def _event_replay_contract(args) -> dict:
if not isinstance(events, list):
raise ValueError('expected JSON array')
except Exception as exc:
candidate_segment_table = _load_1993_candidate_segment_table(args)
return {
'schema': 'jyotish.event_replay_contract.v1', 'status': 'blocked',
'reason': f'event_replay_file_invalid:{exc}', 'events': [],
'candidate_segment_table': candidate_segment_table,
'candidate_segment_reference': (
'candidate_segment_table_not_imported'
if candidate_segment_table
else None
),
'blind_holdout_policy': blind_holdout_policy,
}
candidate_segment_table = _load_1993_candidate_segment_table(args)
return {
'schema': 'jyotish.event_replay_contract.v1',
'status': 'intake_ready',
@@ -2088,21 +2073,10 @@ def _event_replay_contract(args) -> dict:
'Parent/family events may be entered with domain=family and compared against D12 only as '
'candidate discrimination evidence; D12 cannot select a minute by itself.'
),
'candidate_segment_table': candidate_segment_table,
'candidate_segment_reference': (
'candidate_segment_table_not_imported'
if candidate_segment_table
else None
),
'blind_holdout_policy': blind_holdout_policy,
'boundary': 'Events are ingested for later blind replay comparison; intake alone is not a calibration pass.',
}
def _load_1993_candidate_segment_table(args) -> dict | None:
# Author-private candidate table is not imported (hard red line 3 / privacy).
return None
def _load_rectification_evidence_contract() -> dict:
"""Load the versioned event-to-varga registry without scoring personal events."""
path = Path(__file__).resolve().parents[1] / 'references/rectification_evidence_contract_v1.json'
@@ -2120,99 +2094,126 @@ def _load_rectification_evidence_contract() -> dict:
}
DEFAULT_BIRTH_TIME_WINDOW_RADIUS_MINUTES = {"provisional": 15, "approximate": 60}
def _birth_time_accuracy(args) -> str:
missing = object()
value = getattr(args, "birth_time_accuracy", missing)
if value is missing:
return "confirmed"
if not isinstance(value, str) or value not in {"confirmed", "provisional", "approximate"}:
raise ValueError("birth_time_accuracy_must_be_confirmed_provisional_or_approximate")
return value
def _clock_on_birth_date(center: datetime, value) -> datetime | None:
if value is None:
return None
if not isinstance(value, str) or re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", value) is None:
raise ValueError("birth_time_clock_must_be_hh_mm")
parsed = datetime.strptime(value, "%H:%M")
return center.replace(hour=parsed.hour, minute=parsed.minute, second=0, microsecond=0)
def _birth_time_candidate_window(args, center: datetime, accuracy: str) -> tuple[datetime, datetime, datetime]:
candidate_range = getattr(args, "candidate_range", None)
if candidate_range is not None and not isinstance(candidate_range, dict):
raise ValueError("candidate_range_must_be_mapping")
candidate_range = candidate_range or {}
start_raw = candidate_range["start_time"] if "start_time" in candidate_range else getattr(args, "declared_window_start", None)
end_raw = candidate_range["end_time"] if "end_time" in candidate_range else getattr(args, "declared_window_end", None)
has_start = "start_time" in candidate_range or getattr(args, "declared_window_start", None) is not None
has_end = "end_time" in candidate_range or getattr(args, "declared_window_end", None) is not None
representative_raw = (
candidate_range["representative_time"]
if "representative_time" in candidate_range
else getattr(args, "representative_time", None)
)
representative = _clock_on_birth_date(center, representative_raw) or center.replace(second=0, microsecond=0)
if has_start or has_end:
if not has_start or not has_end:
raise ValueError("birth_time_window_requires_start_and_end")
start = _clock_on_birth_date(center, start_raw)
end = _clock_on_birth_date(center, end_raw)
if end < start:
raise ValueError("birth_time_window_must_not_cross_midnight")
else:
before = getattr(args, "uncertainty_before_minutes", None)
after = getattr(args, "uncertainty_after_minutes", None)
radius = DEFAULT_BIRTH_TIME_WINDOW_RADIUS_MINUTES[accuracy]
before = int(before) if isinstance(before, (int, float)) and before >= 0 else radius
after = int(after) if isinstance(after, (int, float)) and after >= 0 else radius
start = max(center.replace(hour=0, minute=0, second=0, microsecond=0), representative - timedelta(minutes=before))
end = min(center.replace(hour=23, minute=59, second=0, microsecond=0), representative + timedelta(minutes=after))
representative = min(max(representative, start), end)
return start, representative, end
def _candidate_minutes(start: datetime, representative: datetime, end: datetime) -> list[str]:
minute_count = int((end - start).total_seconds() // 60) + 1
if minute_count <= 15:
return [(start + timedelta(minutes=index)).strftime("%H:%M") for index in range(minute_count)]
return list(dict.fromkeys(value.strftime("%H:%M") for value in (start, representative, end)))
def _build_birth_time_sensitivity(args) -> dict:
"""Preserve a narrow, unapproved three-minute recast matrix for reporting."""
"""Compare only the trusted unresolved window; never select a birth minute."""
accuracy = _birth_time_accuracy(args)
if accuracy == "confirmed":
return {
"schema": "jyotish.report_birth_time_sensitivity.v1",
"status": "not_applicable",
"accuracy": "confirmed",
}
center = _birth_datetime_from_args(args)
start = center - timedelta(minutes=1)
end = center + timedelta(minutes=1)
start_time = start.strftime('%H:%M')
end_time = end.strftime('%H:%M')
source_reference = {
'review_id': f"rectification-review://report-window/{center.strftime('%Y%m%d-%H%M')}",
'status': 'review_required',
}
approval_gate = _default_birth_time_approval_gate(source_reference)
candidate_segment_table = _load_1993_candidate_segment_table(args)
start, representative, end = _birth_time_candidate_window(args, center, accuracy)
if start == end:
return {
"schema": "jyotish.report_birth_time_sensitivity.v1",
"status": "not_applicable",
"accuracy": "confirmed",
}
candidate_times = _candidate_minutes(start, representative, end)
try:
from flexible_birth_time_profile import build_flexible_birth_time_profile_from_window
from flexible_birth_time_report_support import build_flexible_birth_time_report_support
from flexible_birth_time_full_report_projection import build_flexible_birth_time_full_report_projection
except ImportError: # pragma: no cover - package import
from scripts.flexible_birth_time_profile import build_flexible_birth_time_profile_from_window
from scripts.flexible_birth_time_report_support import build_flexible_birth_time_report_support
from scripts.flexible_birth_time_full_report_projection import build_flexible_birth_time_full_report_projection
profile = build_flexible_birth_time_profile_from_window(
birth_date=center.strftime('%Y-%m-%d'),
start_time=start_time,
end_time=end_time,
lat=float(args.lat),
lon=float(args.lon),
tz=float(args.tz),
ayanamsa=_current_ayanamsa_name(args),
node_mode=getattr(args, 'node_mode', 'mean') or 'mean',
source_reference=source_reference,
)
support = build_flexible_birth_time_report_support(
profile,
approval_gate=approval_gate,
)
projection = build_flexible_birth_time_full_report_projection(support)
return {
'schema': 'jyotish.report_birth_time_sensitivity.v1',
'status': 'candidate_window_only',
'window': {'start': start_time, 'center': center.strftime('%H:%M'), 'end': end_time, 'candidate_count': 3},
'profile': profile,
'approval_gate': approval_gate,
'candidate_segment_table': candidate_segment_table,
'candidate_micro_compare': {
'schema_version': 'jyotish.rectification_candidate_micro_compare.v1',
'status': 'blocked',
'review_required': True,
'candidate_minutes': candidate_segment_table.get('recommended_next_pass', {}).get('candidate_minutes', []) if isinstance(candidate_segment_table, dict) else [],
'leader': None,
'minute_results': [],
'claim_boundary': (
'Minute comparison is packaged for the report but remains blocked until a user event replay is supplied.'
),
},
'micro_compare_minutes': candidate_segment_table.get('recommended_next_pass', {}).get('candidate_minutes', []) if isinstance(candidate_segment_table, dict) else [],
'report_projection': projection,
'required_discrimination_layers': [
'D1', 'D3', 'D4', 'D6', 'D7', 'D9', 'D10', 'D11', 'D12', 'D16', 'D24', 'D30',
'D40', 'D45', 'D60', 'UL', 'A7', 'A10', 'KP cusp',
],
'd12_parent_family_policy': (
'D12 is displayed as stable or minute-sensitive across all three candidates. '
'Only dated parent/family events can use a D12 difference for candidate discrimination; '
'neither D12 nor a single family event can approve a birth minute.'
),
'claim_boundary': (
'This is a local candidate comparison, not a rectification result. It cannot replace the '
'current chart, select a winning minute, or alter Chart Identity.'
),
}
except Exception as exc:
return {
'schema': 'jyotish.report_birth_time_sensitivity.v1',
'status': 'blocked',
'window': {'start': start_time, 'center': center.strftime('%H:%M'), 'end': end_time, 'candidate_count': 3},
'approval_gate': approval_gate,
'candidate_segment_table': candidate_segment_table,
'candidate_micro_compare': {
'schema_version': 'jyotish.rectification_candidate_micro_compare.v1',
'status': 'blocked',
'review_required': True,
'candidate_minutes': candidate_segment_table.get('recommended_next_pass', {}).get('candidate_minutes', []) if isinstance(candidate_segment_table, dict) else [],
'leader': None,
'minute_results': [],
'claim_boundary': (
'Minute comparison is packaged for the report but remains blocked until a user event replay is supplied.'
),
},
'micro_compare_minutes': candidate_segment_table.get('recommended_next_pass', {}).get('candidate_minutes', []) if isinstance(candidate_segment_table, dict) else [],
'reason': f'birth_time_sensitivity_producer_failed:{exc}',
'required_discrimination_layers': [
'D1', 'D3', 'D4', 'D6', 'D7', 'D9', 'D10', 'D11', 'D12', 'D16', 'D24', 'D30',
'D40', 'D45', 'D60', 'UL', 'A7', 'A10', 'KP cusp',
],
}
profile = build_flexible_birth_time_profile_from_window(
birth_date=center.strftime("%Y-%m-%d"),
start_time=start.strftime("%H:%M"),
end_time=end.strftime("%H:%M"),
candidate_times=candidate_times,
lat=float(args.lat),
lon=float(args.lon),
tz=float(args.tz),
ayanamsa=_current_ayanamsa_name(args),
node_mode=getattr(args, "node_mode", "mean") or "mean",
source_reference={
"review_id": f"birth-time-window://{center.strftime('%Y%m%d')}/{start.strftime('%H%M')}-{end.strftime('%H%M')}",
},
)
profile["birth_time_window"]["representative_time"] = representative.strftime("%H:%M")
support = build_flexible_birth_time_report_support(profile)
projection = build_flexible_birth_time_full_report_projection(support)
return {
"schema": "jyotish.report_birth_time_sensitivity.v1",
"status": "candidate_window_only",
"accuracy": accuracy,
"window": projection["window"],
"theme_sensitivity": projection["theme_sensitivity"],
"stable_evidence": projection["stable_structure_section"],
"sensitive_evidence": projection["minute_sensitive_section"],
"report_projection": projection,
"trace": projection["trace"],
"claim_boundary": projection["claim_boundary"],
}
def _default_birth_time_approval_gate(source_reference: dict) -> dict:
@@ -2379,7 +2380,11 @@ def _attach_report_governance_contracts(packet: dict, args) -> dict:
packet['event_replay'] = _event_replay_contract(args)
packet['rectification_evidence_contract'] = _load_rectification_evidence_contract()
packet['raw_module_usage_map'] = _raw_module_usage_map(packet)
packet['birth_time_sensitivity'] = _build_birth_time_sensitivity(args)
birth_time_sensitivity = _build_birth_time_sensitivity(args)
if birth_time_sensitivity.get('status') == 'candidate_window_only':
packet['birth_time_sensitivity'] = birth_time_sensitivity
else:
packet.pop('birth_time_sensitivity', None)
packet['timing_boundary_attribution'] = _build_timing_boundary_attribution(packet)
packet['module_execution_audit'] = _build_module_execution_audit(packet)
ai_pack = ((packet.get('raw_full_reading') or {}).get('ai_prompt_pack') or {})
@@ -9302,91 +9307,12 @@ def render_pl9_markdown(packet: dict) -> str:
lines.append(f"- 边界:{_md_cell(blind_policy.get('boundary'))}")
if event_replay.get('family_d12_binding'):
lines.append(f"- D12/父母家庭绑定:{_md_cell(event_replay.get('family_d12_binding'))}")
candidate_segment_table = event_replay.get('candidate_segment_table') if isinstance(event_replay.get('candidate_segment_table'), dict) else {}
if candidate_segment_table:
summary = candidate_segment_table.get('summary') if isinstance(candidate_segment_table.get('summary'), dict) else {}
primary = summary.get('primary_candidate_window') if isinstance(summary.get('primary_candidate_window'), dict) else {}
if primary:
lines.append(f"- 当前代表时间:{_md_cell(primary.get('representative_time'))}")
lines.append(f"- 当前候选带:{_md_cell(primary.get('start_time'))} - {_md_cell(primary.get('end_time'))}")
lines.append(f"- 校时批准状态:not_approved")
lines.append("- 详细分钟微比较见后文“校时附录”。")
lines.extend(['', '## 生时敏感性矩阵(-', ''])
sensitivity_window = birth_time_sensitivity.get('window') if isinstance(birth_time_sensitivity.get('window'), dict) else {}
lines.append(f"- 状态:{_md_cell(birth_time_sensitivity.get('status', 'blocked'))}")
lines.append(f"- 候选窗口:{_md_cell(sensitivity_window.get('start'))} / {_md_cell(sensitivity_window.get('center'))} / {_md_cell(sensitivity_window.get('end'))}")
approval_gate = birth_time_sensitivity.get('approval_gate') if isinstance(birth_time_sensitivity.get('approval_gate'), dict) else {}
if approval_gate:
lines.append(f"- Review 到 Approval 门槛:{_md_cell(approval_gate.get('status', 'not_provided'))}")
lines.append(f"- 可进入 pending approval{_md_cell(approval_gate.get('can_enter_pending_approval'))}")
if approval_gate.get('blocked_reasons'):
lines.append(f"- 阻断原因:{_md_cell(''.join(str(item) for item in approval_gate.get('blocked_reasons', [])))}")
lines.append(f"- 下一步:{_md_cell(approval_gate.get('required_next_step', 'not_recorded'))}")
lines.append(f"- D12 使用边界:{_md_cell(birth_time_sensitivity.get('d12_parent_family_policy', 'not_recorded'))}")
sensitivity_segment_table = birth_time_sensitivity.get('candidate_segment_table') if isinstance(birth_time_sensitivity.get('candidate_segment_table'), dict) else {}
if sensitivity_segment_table:
lines.append(f"- 候选段表:{_md_cell(sensitivity_segment_table.get('scope', 'candidate_window_only'))}")
sensitivity_micro_compare = birth_time_sensitivity.get('candidate_micro_compare') if isinstance(birth_time_sensitivity.get('candidate_micro_compare'), dict) else {}
if sensitivity_micro_compare:
lines.append(f"- 比较包:{_md_cell(sensitivity_micro_compare.get('status', 'blocked'))}")
sensitivity_profile = birth_time_sensitivity.get('profile') if isinstance(birth_time_sensitivity.get('profile'), dict) else {}
stable_layers = sensitivity_profile.get('stable_evidence') if isinstance(sensitivity_profile.get('stable_evidence'), dict) else {}
sensitive_layers = sensitivity_profile.get('sensitive_evidence') if isinstance(sensitivity_profile.get('sensitive_evidence'), dict) else {}
lines.extend(['', '| 校时层 | 三分钟结果 | 类型 | 使用边界 |', '|--------|------------|------|----------|'])
sensitivity_keys = ('D1.ascendant', 'D3.ascendant', 'D4.ascendant', 'D6.ascendant', 'D7.ascendant', 'D9.ascendant', 'D10.ascendant', 'D11.ascendant', 'D12.ascendant', 'D16.ascendant', 'D24.ascendant', 'D30.ascendant', 'D40.ascendant', 'D45.ascendant', 'D60.ascendant', 'arudha.UL', 'arudha.A7', 'arudha.A10', 'KP.cusp_observation')
for key in sensitivity_keys:
if key in stable_layers:
value, kind = stable_layers.get(key), 'stable_across_window'
elif key in sensitive_layers:
value, kind = sensitive_layers.get(key), 'minute_sensitive'
else:
value, kind = 'not_returned', 'blocked'
boundary = '可作为候选比较的原始结构,不可单独批准出生分钟。'
if key == 'D12.ascendant':
boundary = '仅在有日期化父母/家庭事件时参与候选比较;不得单独选择分钟。'
elif key == 'KP.cusp_observation':
boundary = '只作局部计算观察;exact-cusp parity gate 未闭环。'
lines.append(f"| `{key}` | {_md_cell(_json_safe_report_snapshot(value))} | {kind} | {boundary} |")
lines.append(f"- 总边界:{_md_cell(birth_time_sensitivity.get('claim_boundary', 'not_recorded'))}")
if candidate_segment_table:
lines.extend(['', '### 校时附录:候选分钟计划', ''])
lines.append(f"- 状态:{_md_cell(candidate_segment_table.get('scope', 'candidate_window_only'))}")
lines.append(f"- 结论边界:{_md_cell(candidate_segment_table.get('claim_boundary', 'not_recorded'))}")
summary = candidate_segment_table.get('summary') if isinstance(candidate_segment_table.get('summary'), dict) else {}
primary = summary.get('primary_candidate_window') if isinstance(summary.get('primary_candidate_window'), dict) else {}
if primary:
lines.append(
f"- 主候选:{_md_cell(primary.get('start_time'))} / {_md_cell(primary.get('representative_time'))} / {_md_cell(primary.get('end_time'))}"
)
recommended = candidate_segment_table.get('recommended_next_pass') if isinstance(candidate_segment_table.get('recommended_next_pass'), dict) else {}
if recommended.get('candidate_minutes'):
lines.append(f"- 微调分钟:{_md_cell(' / '.join(str(item) for item in recommended.get('candidate_minutes', [])))}")
if sensitivity_micro_compare:
lines.extend(['', '### 校时附录:分钟排行表', ''])
lines.append(f"- 状态:{_md_cell(sensitivity_micro_compare.get('status', 'blocked'))}")
lines.append(f"- 结论边界:{_md_cell(sensitivity_micro_compare.get('claim_boundary', 'not_recorded'))}")
if sensitivity_micro_compare.get('candidate_minutes'):
lines.append(f"- 候选分钟:{_md_cell(' / '.join(str(item) for item in sensitivity_micro_compare.get('candidate_minutes', [])))}")
if sensitivity_micro_compare.get('leader'):
leader = sensitivity_micro_compare.get('leader') if isinstance(sensitivity_micro_compare.get('leader'), dict) else {}
lines.append(f"- 当前领先:{_md_cell(leader.get('candidate_time'))} / {_md_cell(leader.get('top_score'))}")
ranking_rows = sensitivity_micro_compare.get('minute_results') if isinstance(sensitivity_micro_compare.get('minute_results'), list) else []
if not ranking_rows and sensitivity_micro_compare.get('candidate_minutes'):
ranking_rows = [
{'candidate_time': minute, 'confidence': 'pending_replay', 'top_score': None, 'margin_percent': None, 'note': 'Waiting for event replay'}
for minute in sensitivity_micro_compare.get('candidate_minutes', [])
]
if ranking_rows:
lines.extend(['', '| 排名 | 分钟 | 状态 | 分数 | 边际 | 备注 |', '|------|------|------|------|------|------|'])
for index, row in enumerate(ranking_rows, start=1):
if not isinstance(row, dict):
continue
lines.append(
f"| {index} | {_md_cell(row.get('candidate_time'))} | {_md_cell(row.get('confidence', 'blocked'))} | "
f"{_md_cell(row.get('top_score', 'pending'))} | {_md_cell(row.get('margin_percent', 'pending'))} | "
f"{_md_cell((row.get('candidate_summary') or {}).get('claim_status') or row.get('note') or 'pending')} |"
)
if birth_time_sensitivity.get('status') == 'candidate_window_only':
try:
from flexible_birth_time_report_section import render_flexible_birth_time_report_section
except ImportError: # pragma: no cover - package import
from scripts.flexible_birth_time_report_section import render_flexible_birth_time_report_section
lines.extend(['', render_flexible_birth_time_report_section(birth_time_sensitivity['report_projection']).rstrip()])
lines.extend(['', '## 校时证据领域合同', ''])
lines.append(_md_cell(rectification_evidence_contract.get('claim_boundary', 'rectification_evidence_contract_missing')))
@@ -17375,6 +17301,12 @@ def main():
p.add_argument('--today', default=None, help='Dasha/Sandhi参考日期 YYYY-MM-DD(默认今天)')
p.add_argument('--transit-date', default=None, help='Transit真实过境参考日期 YYYY-MM-DD(默认跟随--today或今天)')
p.add_argument('--target-year', type=int, default=None, help='太阳返照盘目标年份(默认不计算 Varshaphala')
p.add_argument('--birth-time-accuracy', choices=['confirmed', 'provisional', 'approximate'], default='confirmed')
p.add_argument('--declared-window-start', default=None, help='出生时间可信区间开始 HH:MM')
p.add_argument('--declared-window-end', default=None, help='出生时间可信区间结束 HH:MM')
p.add_argument('--representative-time', default=None, help='出生时间可信区间代表分钟 HH:MM')
p.add_argument('--uncertainty-before-minutes', type=int, default=None)
p.add_argument('--uncertainty-after-minutes', type=int, default=None)
p.add_argument('--crosscheck-planets', default=None, help='D4/D9/D10交叉检查行星,逗号分隔')
p.add_argument('--profile-stages', action='store_true', help='输出 full-reading 粗粒度阶段耗时,并在 summary 中附带 stage timings')
+235
View File
@@ -0,0 +1,235 @@
#!/usr/bin/env python3
"""Focused engine/API behavior for report birth-time sensitivity windows."""
from __future__ import annotations
from datetime import datetime
from hashlib import sha256
import json
import os
import sys
from types import SimpleNamespace
import pytest
SCRIPTS = os.path.join(os.path.dirname(__file__), "..", "scripts")
if SCRIPTS not in sys.path:
sys.path.insert(0, SCRIPTS)
import flexible_birth_time_profile as profile_module # noqa: E402
import jyotish_api_server as api # noqa: E402
import jyotish_engine as engine # noqa: E402
def _args(**overrides) -> SimpleNamespace:
values = {
"year": 2004,
"month": 5,
"day": 6,
"hour": 10,
"minute": 7,
"second": 0,
"lat": 12.34,
"lon": 56.78,
"tz": 5.5,
"ayanamsa": "raman",
"node_mode": "mean",
"birth_time_accuracy": "provisional",
"candidate_range": None,
"declared_window_start": None,
"declared_window_end": None,
"representative_time": None,
"uncertainty_before_minutes": None,
"uncertainty_after_minutes": None,
"event_replay_file": None,
}
values.update(overrides)
return SimpleNamespace(**values)
def _fictional_recast(candidate: datetime, **_kwargs) -> dict:
sensitive_sign = "Capricorn" if candidate.minute < 7 else "Aquarius"
return {
"ascendant": {"sign": "Aries"},
"varga_lagna": {
"D9": {"sign": "Taurus"},
"D10": {"sign": sensitive_sign},
},
"arudha": {"A7": {"sign": "Gemini"}, "A10": {"sign": sensitive_sign}, "UL": {"sign": "Cancer"}},
"kp_cusps": {"house_10": {"sub_lord": "Saturn" if candidate.minute < 7 else "Mercury"}},
}
def _digest(value: dict) -> str:
payload = json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
return sha256(payload.encode("utf-8")).hexdigest()
def test_clock_parser_requires_exact_hh_mm() -> None:
center = datetime(2004, 5, 6, 10, 7)
assert engine._clock_on_birth_date(center, "09:05") == datetime(2004, 5, 6, 9, 5)
assert engine._clock_on_birth_date(center, None) is None
for invalid in ("9:05", "09:05:00", "09:05 trailing", "24:00", ""):
with pytest.raises(ValueError, match="birth_time_clock_must_be_hh_mm"):
engine._clock_on_birth_date(center, invalid)
def test_explicit_reverse_or_incomplete_window_is_rejected_instead_of_falling_back() -> None:
center = datetime(2004, 5, 6, 10, 7)
with pytest.raises(ValueError, match="must_not_cross_midnight"):
engine._birth_time_candidate_window(
_args(declared_window_start="23:58", declared_window_end="00:02"),
center,
"provisional",
)
with pytest.raises(ValueError, match="requires_start_and_end"):
engine._birth_time_candidate_window(
_args(declared_window_start="10:00"),
center,
"provisional",
)
with pytest.raises(ValueError, match="birth_time_clock_must_be_hh_mm"):
engine._birth_time_candidate_window(
_args(declared_window_start="10:00 extra", declared_window_end="10:10"),
center,
"provisional",
)
fallback = engine._birth_time_candidate_window(
_args(uncertainty_before_minutes=1, uncertainty_after_minutes=1),
center,
"provisional",
)
assert tuple(value.strftime("%H:%M") for value in fallback) == ("10:06", "10:07", "10:08")
@pytest.mark.parametrize(
"args",
[
_args(birth_time_accuracy="confirmed"),
_args(declared_window_start="10:07", declared_window_end="10:07", representative_time="10:07"),
_args(birth_time_accuracy="approximate", uncertainty_before_minutes=0, uncertainty_after_minutes=0),
],
)
def test_confirmed_or_actual_single_minute_skips_candidate_recast(monkeypatch, args: SimpleNamespace) -> None:
monkeypatch.setattr(
profile_module,
"_recast_candidate_layers",
lambda *_args, **_kwargs: pytest.fail("single-minute input must not recast candidate charts"),
)
assert engine._build_birth_time_sensitivity(args) == {
"schema": "jyotish.report_birth_time_sensitivity.v1",
"status": "not_applicable",
"accuracy": "confirmed",
}
@pytest.mark.parametrize(
("range_status", "end_time", "candidate_count"),
[("accepted", "10:01", 2), ("candidate", "10:14", 15)],
)
def test_multi_minute_accepted_or_candidate_windows_remain_provisional(
monkeypatch, range_status: str, end_time: str, candidate_count: int,
) -> None:
monkeypatch.setattr(profile_module, "_recast_candidate_layers", _fictional_recast)
args = _args(
candidate_range={
"status": range_status,
"start_time": "10:00",
"end_time": end_time,
"representative_time": "10:07",
},
)
first = engine._build_birth_time_sensitivity(args)
second = engine._build_birth_time_sensitivity(args)
assert first["status"] == "candidate_window_only"
assert first["accuracy"] == "provisional"
assert first["window"]["candidate_count"] == candidate_count
assert _digest(first) == _digest(second)
for minute_values in first["sensitive_evidence"]["layers"].values():
assert len({json.dumps(value, sort_keys=True) for value in minute_values.values()}) > 1
for theme in first["theme_sensitivity"].values():
for layer in theme["sensitive_layers"]:
values = first["sensitive_evidence"]["layers"][layer]
assert len({json.dumps(value, sort_keys=True) for value in values.values()}) > 1
def test_confirmed_report_packet_does_not_gain_sensitivity_key() -> None:
packet = {"worksheets": {}, "raw_full_reading": {"modules": {}}}
result = engine._attach_report_governance_contracts(packet, _args(birth_time_accuracy="confirmed"))
assert "birth_time_sensitivity" not in result
@pytest.mark.parametrize("accuracy", ["confirmed", "provisional", "approximate"])
def test_birth_time_accuracy_accepts_only_exact_supported_values(accuracy: str) -> None:
assert engine._birth_time_accuracy(_args(birth_time_accuracy=accuracy)) == accuracy
def test_birth_time_accuracy_defaults_to_confirmed_when_field_is_absent() -> None:
assert engine._birth_time_accuracy(SimpleNamespace()) == "confirmed"
@pytest.mark.parametrize("invalid", [None, "", "bogus", " PROVISIONAL ", "PROVISIONAL", 1])
def test_birth_time_accuracy_rejects_invalid_or_formatted_values(invalid: object) -> None:
with pytest.raises(
ValueError,
match="birth_time_accuracy_must_be_confirmed_provisional_or_approximate",
):
engine._birth_time_accuracy(_args(birth_time_accuracy=invalid))
def test_api_turns_invalid_birth_time_accuracy_into_bad_request(monkeypatch) -> None:
class FakeEngine:
@staticmethod
def cmd_full_reading(args):
engine._birth_time_accuracy(args)
return {"modules": {}}
monkeypatch.setattr(api, "_load_local_module", lambda _name: FakeEngine)
handler = api.JyotishAPIHandler.__new__(api.JyotishAPIHandler)
with pytest.raises(
api.BadRequest,
match="birth_time_accuracy_must_be_confirmed_provisional_or_approximate",
):
handler._compute_full_reading_for_thematic({
"year": 2004,
"month": 5,
"day": 6,
"hour": 10,
"minute": 7,
"lat": 12.34,
"lon": 56.78,
"tz": 5.5,
"birth_time_accuracy": " PROVISIONAL ",
})
def test_api_turns_invalid_report_window_into_bad_request(monkeypatch) -> None:
class FakeEngine:
@staticmethod
def cmd_full_reading(_args):
raise ValueError("birth_time_window_must_not_cross_midnight")
monkeypatch.setattr(api, "_load_local_module", lambda _name: FakeEngine)
handler = api.JyotishAPIHandler.__new__(api.JyotishAPIHandler)
with pytest.raises(api.BadRequest, match="birth_time_window_must_not_cross_midnight"):
handler._compute_full_reading_for_thematic({
"year": 2004,
"month": 5,
"day": 6,
"hour": 10,
"minute": 7,
"lat": 12.34,
"lon": 56.78,
"tz": 5.5,
"birth_time_accuracy": "provisional",
"declared_window_start": "23:58",
"declared_window_end": "00:02",
})
+86
View File
@@ -0,0 +1,86 @@
#!/usr/bin/env python3
"""Focused contracts for unresolved birth-time candidate profiles."""
from __future__ import annotations
import json
import os
import sys
import pytest
SCRIPTS = os.path.join(os.path.dirname(__file__), "..", "scripts")
if SCRIPTS not in sys.path:
sys.path.insert(0, SCRIPTS)
from flexible_birth_time_profile import ( # noqa: E402
FlexibleBirthTimeProfileError,
build_flexible_birth_time_profile,
)
def _candidate(index: int, count: int) -> dict:
candidate_time = f"08:{index:02d}"
return {
"candidate_id": f"fictional-{index}",
"candidate_time": candidate_time,
"evidence": {
"D1.ascendant": "Aries",
"D10.ascendant": "Capricorn" if index < count - 1 else "Aquarius",
},
"trace": [{"kind": "fictional_candidate", "reference": f"fixture://{candidate_time}"}],
}
@pytest.mark.parametrize("count", [2, 15])
def test_profile_supports_two_to_fifteen_candidates_with_real_minute_variation(count: int) -> None:
candidates = [_candidate(index, count) for index in range(count)]
profile = build_flexible_birth_time_profile(
candidates,
source_reference={"review_id": "fictional-review"},
)
reversed_profile = build_flexible_birth_time_profile(
list(reversed(candidates)),
source_reference={"review_id": "fictional-review"},
)
assert profile == reversed_profile
assert profile["birth_time_window"]["candidate_count"] == count
assert profile["stable_evidence"] == {"D1.ascendant": "Aries"}
assert set(profile["sensitive_evidence"]) == {"D10.ascendant"}
for minute_values in profile["sensitive_evidence"].values():
assert len({json.dumps(value, sort_keys=True) for value in minute_values.values()}) > 1
@pytest.mark.parametrize("count", [1, 16])
def test_profile_rejects_candidate_counts_outside_two_to_fifteen(count: int) -> None:
with pytest.raises(FlexibleBirthTimeProfileError, match="candidate_count_must_be_two_to_fifteen"):
build_flexible_birth_time_profile(
[_candidate(index, max(count, 2)) for index in range(count)],
source_reference={"review_id": "fictional-review"},
)
@pytest.mark.parametrize("status", ["approved", "confirmed"])
def test_profile_rejects_approved_or_confirmed_source_reference(status: str) -> None:
with pytest.raises(FlexibleBirthTimeProfileError, match="approved_or_confirmed_source_reference_forbidden"):
build_flexible_birth_time_profile(
[_candidate(0, 2), _candidate(1, 2)],
source_reference={"review_id": "fictional-review", "status": status},
)
@pytest.mark.parametrize(
"field",
["approved_birth_time", "final_birth_time", "winner", "approval", "approval_authority"],
)
def test_profile_rejects_birth_minute_authority_fields(field: str) -> None:
candidates = [_candidate(0, 2), _candidate(1, 2)]
candidates[0]["evidence"][field] = "08:00"
with pytest.raises(FlexibleBirthTimeProfileError, match="candidate_window_authority_forbidden"):
build_flexible_birth_time_profile(
candidates,
source_reference={"review_id": "fictional-review"},
)
@@ -0,0 +1,95 @@
#!/usr/bin/env python3
"""Negative-authority contracts for flexible birth-time report projections."""
from __future__ import annotations
from copy import deepcopy
import os
import sys
import pytest
SCRIPTS = os.path.join(os.path.dirname(__file__), "..", "scripts")
if SCRIPTS not in sys.path:
sys.path.insert(0, SCRIPTS)
from flexible_birth_time_full_report_projection import ( # noqa: E402
FlexibleBirthTimeFullReportProjectionError,
build_flexible_birth_time_full_report_projection,
)
from flexible_birth_time_profile import build_flexible_birth_time_profile # noqa: E402
from flexible_birth_time_report_section import ( # noqa: E402
FlexibleBirthTimeReportSectionError,
render_flexible_birth_time_report_section,
)
from flexible_birth_time_report_support import ( # noqa: E402
FlexibleBirthTimeReportSupportError,
build_flexible_birth_time_report_support,
)
def _profile() -> dict:
rows = [
{
"candidate_id": "fictional-a",
"candidate_time": "08:00",
"evidence": {"D1.ascendant": "Aries", "D10.ascendant": "Capricorn"},
"trace": [{"kind": "fictional_candidate", "reference": "fixture://08:00"}],
},
{
"candidate_id": "fictional-b",
"candidate_time": "08:01",
"evidence": {"D1.ascendant": "Aries", "D10.ascendant": "Aquarius"},
"trace": [{"kind": "fictional_candidate", "reference": "fixture://08:01"}],
},
]
profile = build_flexible_birth_time_profile(rows, source_reference={"review_id": "fictional-review"})
profile["birth_time_window"]["representative_time"] = "08:00"
return profile
def _packets() -> tuple[dict, dict, dict]:
profile = _profile()
support = build_flexible_birth_time_report_support(profile)
projection = build_flexible_birth_time_full_report_projection(support)
return profile, support, projection
def test_report_projection_is_hash_stable_and_renders_only_candidate_window_evidence() -> None:
_, support, projection = _packets()
reordered = deepcopy(support)
reordered["birth_time_window"] = dict(reversed(list(reordered["birth_time_window"].items())))
assert build_flexible_birth_time_full_report_projection(reordered)["projection_id"] == projection["projection_id"]
assert projection["status"] == "candidate_window_only"
assert "出生时间敏感度" in render_flexible_birth_time_report_section(projection)
@pytest.mark.parametrize("field", ["approved_birth_time", "final_birth_time", "winner", "approval"])
def test_every_report_layer_rejects_birth_minute_authority_fields(field: str) -> None:
profile, support, projection = _packets()
cases = [
(profile, build_flexible_birth_time_report_support, FlexibleBirthTimeReportSupportError),
(support, build_flexible_birth_time_full_report_projection, FlexibleBirthTimeFullReportProjectionError),
(projection, render_flexible_birth_time_report_section, FlexibleBirthTimeReportSectionError),
]
for packet, consumer, error_type in cases:
tampered = deepcopy(packet)
tampered["authority_probe"] = {field: "08:00"}
with pytest.raises(error_type, match="candidate_window_authority_forbidden"):
consumer(tampered)
@pytest.mark.parametrize("status", ["approved", "confirmed"])
def test_every_report_layer_rejects_approved_or_confirmed_source_reference(status: str) -> None:
profile, support, projection = _packets()
cases = [
(profile, build_flexible_birth_time_report_support, FlexibleBirthTimeReportSupportError),
(support, build_flexible_birth_time_full_report_projection, FlexibleBirthTimeFullReportProjectionError),
(projection, render_flexible_birth_time_report_section, FlexibleBirthTimeReportSectionError),
]
for packet, consumer, error_type in cases:
tampered = deepcopy(packet)
tampered["source_reference"] = {"review_id": "fictional-review", "status": status}
with pytest.raises(error_type, match="candidate_window_authority_forbidden"):
consumer(tampered)