Skip to content

Use idiomatic Effect schemas and random in server flows #2815

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
cursor wants to merge 3 commits into
base: main
Choose a base branch
from
+148 −78
Draft

Use idiomatic Effect schemas and random in server flows #2815

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
da81532
Use Effect schemas and random in server flows
cursoragent May 26, 2026
c7683bc
Format trace diagnostics helpers
cursoragent May 26, 2026
e1e67e2
Fix trace diagnostics event decoding
cursoragent May 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
156 changes: 87 additions & 69 deletions apps/server/src/diagnostics/TraceDiagnostics.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,23 +14,32 @@ import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as PlatformError from "effect/PlatformError";
import * as Schema from "effect/Schema";

const TraceRecordLine = Schema.Struct({
name: Schema.optional(Schema.Unknown),
traceId: Schema.optional(Schema.Unknown),
spanId: Schema.optional(Schema.Unknown),
startTimeUnixNano: Schema.optional(Schema.Unknown),
endTimeUnixNano: Schema.optional(Schema.Unknown),
durationMs: Schema.optional(Schema.Unknown),
exit: Schema.optional(Schema.Unknown),
events: Schema.optional(Schema.Array(Schema.Unknown)),
});

interface TraceRecordLike {
readonly name?: unknown;
readonly traceId?: unknown;
readonly spanId?: unknown;
readonly startTimeUnixNano?: unknown;
readonly endTimeUnixNano?: unknown;
readonly durationMs?: unknown;
readonly exit?: unknown;
readonly events?: unknown;
}
const TraceEvent = Schema.Struct({
name: Schema.optional(Schema.Unknown),
timeUnixNano: Schema.optional(Schema.Unknown),
attributes: Schema.optional(Schema.Unknown),
});

interface TraceEventLike {
readonly name?: unknown;
readonly timeUnixNano?: unknown;
readonly attributes?: unknown;
}
const TraceEventAttributes = Schema.Record(Schema.String, Schema.Unknown);

type TraceEventLike = typeof TraceEvent.Type;

const decodeTraceRecordLine = Schema.decodeUnknownOption(Schema.fromJsonString(TraceRecordLine));
const decodeTraceEvent = Schema.decodeUnknownOption(TraceEvent);
const decodeTraceEventAttributes = Schema.decodeUnknownOption(TraceEventAttributes);

export interface TraceDiagnosticsOptions {
readonly traceFilePath: string;
Expand Down Expand Up @@ -74,47 +83,42 @@ function toRotatedTracePaths(traceFilePath: string, maxFiles: number): ReadonlyA
return [...backups, traceFilePath];
}

function isRecordObject(value: unknown): value is TraceRecordLike {
function isRecordObject(value: unknown): value is Readonly<Record<string, unknown>> {
return typeof value === "object" && value !== null;
}

function toStringValue(value: unknown): string | null {
return typeof value === "string" && value.trim().length > 0 ? value : null;
function toStringValue(value: unknown): Option.Option<string> {
return typeof value === "string" && value.trim().length > 0 ? Option.some(value) : Option.none();
}

function toNumberValue(value: unknown): number | null {
return typeof value === "number" && Number.isFinite(value) ? value : null;
function toNumberValue(value: unknown): Option.Option<number> {
return typeof value === "number" && Number.isFinite(value) ? Option.some(value) : Option.none();
}

function unixNanoToDateTime(value: unknown): DateTime.Utc | null {
function unixNanoToDateTime(value: unknown): Option.Option<DateTime.Utc> {
const text = toStringValue(value);
if (!text) return null;
if (Option.isNone(text)) return Option.none();
try {
const millis = Number(BigInt(text) / 1_000_000n);
return Option.getOrNull(DateTime.make(millis));
const millis = Number(BigInt(text.value) / 1_000_000n);
return DateTime.make(millis);
} catch {
return null;
return Option.none();
}
}

function readExitTag(exit: unknown): string | null {
if (!isRecordObject(exit) || !("_tag" in exit)) return null;
function readExitTag(exit: unknown): Option.Option<string> {
if (!isRecordObject(exit) || !("_tag" in exit)) return Option.none();
return toStringValue(exit._tag);
}

function readExitCause(exit: unknown): string {
if (!isRecordObject(exit) || !("cause" in exit)) return "Failure";
return toStringValue(exit.cause)?.trim() ?? "Failure";
}

function isTraceEvent(value: unknown): value is TraceEventLike {
return typeof value === "object" && value !== null;
const cause = toStringValue(exit.cause);
return Option.isSome(cause) ? cause.value.trim() : "Failure";
}

function readEventAttributes(event: TraceEventLike): Readonly<Record<string, unknown>> {
return typeof event.attributes === "object" && event.attributes !== null
? (event.attributes as Readonly<Record<string, unknown>>)
: {};
return Option.getOrElse(decodeTraceEventAttributes(event.attributes), () => ({}));
}

function makeEmptyDiagnostics(input: {
Expand Down Expand Up @@ -199,8 +203,8 @@ export function aggregateTraceDiagnostics(
let failureCount = 0;
let interruptionCount = 0;
let slowSpanCount = 0;
let firstSpanAt: DateTime.Utc | null = null;
let lastSpanAt: DateTime.Utc | null = null;
let firstSpanAt = Option.none<DateTime.Utc>();
let lastSpanAt = Option.none<DateTime.Utc>();

const spansByName = new Map<
string,
Expand All @@ -217,42 +221,50 @@ export function aggregateTraceDiagnostics(
for (const line of lines) {
if (line.trim().length === 0) continue;

let parsed: unknown;
try {
parsed = JSON.parse(line);
} catch {
const parsedOption = decodeTraceRecordLine(line);
if (Option.isNone(parsedOption)) {
parseErrorCount += 1;
continue;
}
const parsed = parsedOption.value;

if (!isRecordObject(parsed)) {
parseErrorCount += 1;
continue;
}

const name = toStringValue(parsed.name);
const traceId = toStringValue(parsed.traceId);
const spanId = toStringValue(parsed.spanId);
const durationMs = toNumberValue(parsed.durationMs);
const endedAt = unixNanoToDateTime(parsed.endTimeUnixNano);
const nameOption = toStringValue(parsed.name);
const traceIdOption = toStringValue(parsed.traceId);
const spanIdOption = toStringValue(parsed.spanId);
const durationMsOption = toNumberValue(parsed.durationMs);
const endedAtOption = unixNanoToDateTime(parsed.endTimeUnixNano);
const startedAt = unixNanoToDateTime(parsed.startTimeUnixNano);

if (!name || !traceId || !spanId || durationMs === null || !endedAt) {
if (
Option.isNone(nameOption) ||
Option.isNone(traceIdOption) ||
Option.isNone(spanIdOption) ||
Option.isNone(durationMsOption) ||
Option.isNone(endedAtOption)
) {
parseErrorCount += 1;
continue;
}
const name = nameOption.value;
const traceId = traceIdOption.value;
const spanId = spanIdOption.value;
const durationMs = durationMsOption.value;
const endedAt = endedAtOption.value;

recordCount += 1;
firstSpanAt =
startedAt && (firstSpanAt === null || DateTime.isLessThan(startedAt, firstSpanAt))
? startedAt
: firstSpanAt;
lastSpanAt =
lastSpanAt === null || DateTime.isGreaterThan(endedAt, lastSpanAt) ? endedAt : lastSpanAt;
if (
Option.isSome(startedAt) &&
(Option.isNone(firstSpanAt) || DateTime.isLessThan(startedAt.value, firstSpanAt.value))
) {
firstSpanAt = startedAt;
}
if (Option.isNone(lastSpanAt) || DateTime.isGreaterThan(endedAt, lastSpanAt.value)) {
lastSpanAt = Option.some(endedAt);
}

const exitTag = readExitTag(parsed.exit);
const isFailure = exitTag === "Failure";
const isInterrupted = exitTag === "Interrupted";
const isFailure = Option.isSome(exitTag) && exitTag.value === "Failure";
const isInterrupted = Option.isSome(exitTag) && exitTag.value === "Interrupted";
if (isFailure) failureCount += 1;
if (isInterrupted) interruptionCount += 1;

Expand Down Expand Up @@ -292,11 +304,14 @@ export function aggregateTraceDiagnostics(
}

if (Array.isArray(parsed.events)) {
for (const rawEvent of parsed.events) {
if (!isTraceEvent(rawEvent)) continue;
const attributes = readEventAttributes(rawEvent);
const level = toStringValue(attributes["effect.logLevel"]);
if (!level) continue;
for (const rawTraceEvent of parsed.events) {
const eventOption = decodeTraceEvent(rawTraceEvent);
if (Option.isNone(eventOption)) continue;
const event = eventOption.value;
const attributes = readEventAttributes(event);
const levelOption = toStringValue(attributes["effect.logLevel"]);
if (Option.isNone(levelOption)) continue;
const level = levelOption.value;

logLevelCounts[level] = (logLevelCounts[level] ?? 0) + 1;
const normalizedLevel = level.toLowerCase();
Expand All @@ -309,8 +324,11 @@ export function aggregateTraceDiagnostics(
continue;
}

const seenAt = unixNanoToDateTime(rawEvent.timeUnixNano) ?? endedAt;
const message = toStringValue(rawEvent.name)?.trim() ?? "Log event";
const seenAt = Option.getOrElse(unixNanoToDateTime(event.timeUnixNano), () => endedAt);
const message = Option.getOrElse(
Option.map(toStringValue(event.name), (value) => value.trim()),
() => "Log event",
);
latestWarningAndErrorLogs.push({
spanName: name,
level,
Expand Down Expand Up @@ -342,8 +360,8 @@ export function aggregateTraceDiagnostics(
readAt,
recordCount,
parseErrorCount,
firstSpanAt: Option.fromNullishOr(firstSpanAt),
lastSpanAt: Option.fromNullishOr(lastSpanAt),
firstSpanAt,
lastSpanAt,
failureCount,
interruptionCount,
slowSpanThresholdMs,
Expand Down
54 changes: 53 additions & 1 deletion apps/server/src/git/GitManager.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,12 @@ import path from "node:path";
import { spawnSync } from "node:child_process";

import * as NodeServices from "@effect/platform-node/NodeServices";
import { it } from "@effect/vitest";
import { assert, it } from "@effect/vitest";
import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as PlatformError from "effect/PlatformError";
import * as Random from "effect/Random";
import * as Scope from "effect/Scope";
import { ChildProcessSpawner } from "effect/unstable/process";
import { expect } from "vitest";
Expand Down Expand Up @@ -188,6 +189,22 @@ function runGitSyncForFakeGh(cwd: string, args: readonly string[]): void {
});
}

function makeDeterministicRandomService(seed = 0x1234_5678): {
nextIntUnsafe: () => number;
nextDoubleUnsafe: () => number;
} {
let state = seed >>> 0;
const nextIntUnsafe = (): number => {
state = (Math.imul(1_664_525, state) + 1_013_904_223) >>> 0;
return state;
};

return {
nextIntUnsafe,
nextDoubleUnsafe: () => nextIntUnsafe() / 0x1_0000_0000,
};
}

function isGitHubCliError(error: unknown): error is GitHubCliError {
return (
typeof error === "object" &&
Expand Down Expand Up @@ -3213,6 +3230,41 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => {
}),
);

it.effect("uses Effect Random for implicit progress action ids", () =>
Effect.gen(function* () {
const repoDir = yield* makeTempDir("t3code-git-manager-");
yield* initRepo(repoDir);
fs.writeFileSync(path.join(repoDir, "random-action-id.txt"), "random action id\n");

const { manager } = yield* makeManager();
const events: GitActionProgressEvent[] = [];
const seed = 0x5eed_c0de;
const expectedActionId = yield* Random.nextUUIDv4.pipe(
Effect.provideService(Random.Random, makeDeterministicRandomService(seed)),
);

const result = yield* runStackedAction(
manager,
{
cwd: repoDir,
action: "commit",
},
{
progressReporter: {
publish: (event) =>
Effect.sync(() => {
events.push(event);
}),
},
},
).pipe(Effect.provideService(Random.Random, makeDeterministicRandomService(seed)));

assert.equal(result.commit.status, "created");
assert.isAbove(events.length, 0);
assert.isTrue(events.every((event) => event.actionId === expectedActionId));
}),
);

it.effect("emits ordered progress events for commit hooks", () =>
Effect.gen(function* () {
const repoDir = yield* makeTempDir("t3code-git-manager-");
Expand Down
16 changes: 8 additions & 8 deletions apps/server/src/git/GitManager.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,3 @@
import { randomUUID } from "node:crypto";

import * as Arr from "effect/Array";
import * as Cache from "effect/Cache";
import * as Context from "effect/Context";
Expand All @@ -12,6 +10,7 @@ import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Order from "effect/Order";
import * as Path from "effect/Path";
import * as Random from "effect/Random";
import * as Ref from "effect/Ref";
import {
GitActionProgressEvent,
Expand Down Expand Up @@ -536,11 +535,11 @@ export const makeGitManager = Effect.fn("makeGitManager")(function* () {
const sourceControlProvider = (cwd: string) => sourceControlProviders.resolve({ cwd });
const serverSettingsService = yield* ServerSettingsService;

const createProgressEmitter = (
const createProgressEmitter = Effect.fn("GitManager.createProgressEmitter")(function* (
input: { cwd: string; action: GitStackedAction },
options?: GitRunStackedActionOptions,
) => {
const actionId = options?.actionId ?? randomUUID();
) {
const actionId = options?.actionId ?? (yield* Random.nextUUIDv4);
const reporter = options?.progressReporter;

const emit = (event: GitActionProgressPayload) =>
Expand All @@ -557,7 +556,7 @@ export const makeGitManager = Effect.fn("makeGitManager")(function* () {
actionId,
emit,
};
};
});

const configurePullRequestHeadUpstreamBase = Effect.fn("configurePullRequestHeadUpstream")(
function* (
Expand Down Expand Up @@ -1301,7 +1300,8 @@ export const makeGitManager = Effect.fn("makeGitManager")(function* () {
modelSelection,
});

const bodyFile = path.join(tempDir, `t3code-pr-body-${process.pid}-${randomUUID()}.md`);
const bodyFileId = yield* Random.nextUUIDv4;
const bodyFile = path.join(tempDir, `t3code-pr-body-${process.pid}-${bodyFileId}.md`);
yield* fileSystem
.writeFileString(bodyFile, generated.body)
.pipe(
Expand Down Expand Up @@ -1591,7 +1591,7 @@ export const makeGitManager = Effect.fn("makeGitManager")(function* () {

const runStackedAction: GitManagerShape["runStackedAction"] = Effect.fn("runStackedAction")(
function* (input, options) {
const progress = createProgressEmitter(input, options);
const progress = yield* createProgressEmitter(input, options);
const currentPhase = yield* Ref.make<Option.Option<GitActionProgressPhase>>(Option.none());

const runAction = Effect.fn("runStackedAction.runAction")(function* (): Effect.fn.Return<
Expand Down
Loading