mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 04:02:18 +01:00
Trace native preparation and OpenCode execution phases (#17918)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Each native run prepares its workspace and skills, then starts its provider. > - A long preparation span does not show whether inventory refresh, a cache lock, repository reads, or file checks caused the delay. > - One hosted run spent 81 seconds in this phase. The same deployed service and pinned files took 2.55 seconds in a separate diagnostic process. > - Another hosted turn spent 69 seconds before its first output. A matched public-workspace local model replay took 7.64 seconds. > - We need skill preparation and provider-internal measurements before choosing another fix. > - This pull request adds skill spans, a real-service benchmark, and opt-in filtered provider timing. > - The benefit is a faster and more precise diagnosis loop. ## Linked Issues or Issue Description Refs #17620. That PR owns the batched read optimization. This PR adds diagnostics and does not change its read policy. Refs #17907. That PR protects long cache lock transactions from the database idle timeout. **What happened?** Native skill configuration can take much longer than matched standalone resolution. Provider execution also has long gaps with no internal timings. Existing spans cannot separate inventory refresh, cache acquisition, repository reads, context loading, Git snapshots, or model streaming. **Expected behavior** An operator-configured trace shows these phases under the existing run context. Without an OTLP endpoint, the new spans stay off. A local benchmark can reproduce the service path with real PostgreSQL, files, an agent actor, and a responsible user. **Steps to reproduce** 1. Run `pnpm --filter @paperclipai/server exec tsx scripts/benchmark-native-skill-preparation.ts --actor=agent --staging-shape=true`. 2. Inspect cold and warm phase timings, SQL counts, and file verification results. 3. Set the existing operator OTLP configuration and add `--export-otlp=true` to inspect the same hierarchy in the collector. ## What Changed - Add bounded skill inventory, refresh wait, cache acquisition, row, version, repository, file check, and consumer spans. - Keep the existing endpoint gate, attribute allowlist, cache locks, authorization, file validation, and durable pin boundary. - Add a standalone benchmark with a full repository fixture, immutable snapshot input, calibrated local transport delay, and optional OTLP export. - Verify every delivered file and executable bit outside the timed section. Create its fixture agent through the lifecycle API. - Add an opt-in authenticated loopback sink for OpenCode. Discard raw attributes, logs and content. Retain only closed phase names and timestamps. - Carry one bounded diagnostic batch through the real proxy, Rust normalization, durable journal and server trace projection. Group provider phases under their execution parent. - Freeze the Date clock in an existing delegated Slack rate-limit test. This fixes a CI failure when consecutive requests cross a rate-window boundary. It does not change rate-limit behavior. - Document phase boundaries and the difference between SDK startup and confirmed ingestion. ## Verification - Skill service, runtime selection, concurrency, cache and timing suites: 122 tests passed with real PostgreSQL and migrations. Enabled-span integration tests also passed. They verify parent relationships and a real advisory-lock timeout. - A captured 208-file repository produced 209 verified runtime files. The benchmark uses real agent and responsible-user authorization. The lifecycle fixture correction passed with the staging-shaped inventory. - Affected OpenCode driver, collector and transport tests: 72 passed. The tests cover authentication, content removal, size/count bounds, disabled behavior and the rebuilt native Runner boundary. One Rust projection test and 18 server trace tests also passed. A held run-log sink proves that a full 128-span diagnostic batch adds no duplicate database writes before terminal consumption. - The two CI-failing delegated Slack rate-limit cases passed against real PostgreSQL with only the Date clock frozen. - Runner TypeScript and server TypeScript checks passed. Rust formatting and module-boundary checks passed. - Real Honeycomb readback confirmed the 14-span skill waterfall and a 40-span provider waterfall. Provider phases have the correct execution parent. The provider export uses synthetic model replies and is instrumentation proof only. - Earlier full local checks failed on missing package-local executables, then on the default Node heap and a missing generated service worker. The dependency links were repaired. The server check passed with a larger heap. The full local test attempt could not start embedded PostgreSQL on this host and was stopped after retaining its failures; real database coverage uses an isolated database on an existing local server. These attempts do not count as a full local pass. - CI on the prior revision passed typecheck, build, Runner and all E2E shards. It failed the benchmark lifecycle boundary and one existing rate-window test. Those fixes passed CI. A later Runner assertion exposed the Copilot source identity pin on the shared transport file. The diagnostic flag now uses the existing configured-environment projection, preserving that file byte-for-byte. All 66 selected installation and native transport tests passed. New-head CI and review remain required. - The under-15-second hosted response target remains unmet. This PR adds diagnostics and does not claim a hosted latency fix. ## Risks - Skill spans and Honeycomb export retain the operator endpoint gate. OpenCode phase collection requires a separate explicit diagnostic flag. Without an endpoint, its closed measurements stay in the instance run log. - Raw OpenCode OTLP can contain prompts and authorization attributes. The new sink rejects unauthenticated requests and discards all attributes before any measurement reaches PRP or the operator collector. Request, event and per-phase limits bound memory and journal growth. - Collection does not wait for a provider flush. Late or oversized exports can be absent; these measurements are not a complete profiling guarantee. - The benchmark's inventory contents and transport delay are explicit fixtures. They do not reproduce server contention or prove hosted response time. - The one durable diagnostic batch remains the local evidence. Its child span export does not add per-phase run-log rows. - No database migration, authorization policy, model selection or inference transport changes. The Rust Runner and OpenCode proxy must be shipped together for the new opt-in diagnostic. ## Model Used OpenAI GPT-6 through Codex, with repository inspection, shell execution, browser tools, and Honeycomb queries. The session does not expose a verifiable backend deployment ID or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Paperclip <paperclip@paperclip.ing>
This commit is contained in:
24 files changed
+930
-19
No files matched your search
@@ -838,6 +838,41 @@ Set `SEARCH_EVAL_SCALE=1` to additionally measure a disposable 10,000-task,
|
||||
30,000-comment dataset. `SEARCH_EVAL_REPORT=/tmp/search-quality.json` saves
|
||||
per-query results and latency measurements; scale measurements are opt-in.
|
||||
|
||||
### Native skill preparation benchmark
|
||||
|
||||
Measure native skill preparation through the production service, real PostgreSQL,
|
||||
and an in-memory OpenTelemetry exporter:
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/server exec tsx scripts/benchmark-native-skill-preparation.ts
|
||||
```
|
||||
|
||||
The default fixture has a 208-file repository of about 10.8 MB. The report includes
|
||||
cold and warm resolution, a fresh service instance, detailed phase timings, and
|
||||
regular-pool SQL statement counts (excluding the separate cache-lock connection).
|
||||
Every delivered file is checked outside the timed section. No provider is started.
|
||||
`--snapshot-file=/absolute/path/snapshot.json --entrypoint=skills/example/SKILL.md`
|
||||
uses a captured repository file inventory instead. `--network-delay-ms=25` adds
|
||||
25 ms in each direction through a local TCP relay; this is simulated transport
|
||||
latency, not a measurement of a hosted database.
|
||||
|
||||
`--actor=agent` adds a real agent, run, and responsible user, then records version
|
||||
pins through the same callback boundary as execution. `--staging-shape=true`
|
||||
seeds 14 projects, 10 sources, 8 bindings, and 156 other skills. These match one
|
||||
observed staging inventory's counts; their content is synthetic. Combine these
|
||||
options to include authorization and unselected library rows in the measurement.
|
||||
|
||||
By default, spans stay in memory. `--export-otlp=true` also exports them using
|
||||
the operator's `OTEL_EXPORTER_OTLP_ENDPOINT`, protocol, and headers. Keep header
|
||||
values out of command arguments and reports. The report records source hashes,
|
||||
the export mode, and trace IDs. Exported traces identify this local diagnostic
|
||||
in their service version; they are not evidence of staging performance.
|
||||
|
||||
If creating another embedded cluster is impractical, `--postgres-port=55448
|
||||
--postgres-user=paperclip` uses an existing **local** server. It creates and migrates
|
||||
a uniquely named disposable database and drops that database afterward; it does
|
||||
not reuse an application database.
|
||||
|
||||
### Repository workspace admission benchmark
|
||||
|
||||
Measure a fresh checkout with and without the repository object cache:
|
||||
|
||||
@@ -35,6 +35,13 @@ maintains as a dependency; once you install the packages below, they become
|
||||
normal dependencies of **your own** project, and your own Dependabot updates
|
||||
them.
|
||||
|
||||
An initialized SDK does not confirm ingestion. Verify a fresh trace from the
|
||||
deployed revision at the configured destination. A rejected export can leave
|
||||
the application healthy. For Honeycomb, configure the ingest key in the
|
||||
`x-honeycomb-team` entry of `OTEL_EXPORTER_OTLP_HEADERS`; see
|
||||
[Honeycomb authentication](https://docs.honeycomb.io/api/query-annotations/authorization).
|
||||
Keep credential values out of logs and diagnostic reports.
|
||||
|
||||
## Budget admission, accounting, and enforcement
|
||||
|
||||
The operator-gated `paperclip.budgets` tracer records
|
||||
@@ -313,6 +320,9 @@ attempt or owning phase:
|
||||
| `runtime_selection.*` | Completion contracts, session/recovery reads, profiles, interaction materialization, backend discovery, bootstrap, run/coordinator writes, transaction entry/settlement, and workspace synchronization. |
|
||||
| `runtime_context.instructions`, `.skills`, `.mcp` | The three concurrent runtime-context branches. Their durations overlap. |
|
||||
| `runtime_context.asset_*` | Source reads, manifest hashing, cache lookup, cold materialization, and exact file verification on cold and warm paths. |
|
||||
| `skills.inventory`, `.wait`, `.bundled`, `.releases`, `.folders`, `.local_sources` | Company skill reconciliation, waiting for an existing company refresh, and its bundled imports, release verification, folder, and local-source work beneath runtime skill preparation. Inventory children overlap their parent. |
|
||||
| `skills.cache.acquire`, `skills.rows`, `skills.version.read`, `skills.repository.read`, `skills.version.materialize`, `skills.version.verify` | Dedicated connection/transaction startup plus advisory-lock acquisition, authorized skill rows, frozen version and repository reads, and exact runtime materialization/validation. Lock acquisition ends before the protected work; version verification is inside materialization. Only bounded file counts are added. |
|
||||
| `skills.consume` | The caller's delivery and durable version-pinning callback while the cache lock remains held. |
|
||||
| `workspace_sync.prepare_host_current`, `.prepare_durable_seed`, `.prepare_adopt_remote` | Runtime preparation under the selected inbound mode, including snapshot and staging work. A warm mismatch may produce a second preparation; it is real repeated work. |
|
||||
| `workspace_sync.source_root`, `.repositories`, `.ignore_rules`, `.snapshot_git`, `.snapshot_baseline` | Physical source capture, managed repositories, ignore rules, Git metadata and baseline content snapshot. |
|
||||
| `workspace_sync.pack`, `.git_clone`, `.git_archive`, `.overlay_copy`, `.overlay_archive`, `.deleted_manifest` | Temporary shallow Git clone, its archive callback, selected working-tree overlay copy/archive, and the deleted-file manifest. The archive is inside the clone's lifetime; do not sum parent and child. |
|
||||
@@ -334,6 +344,33 @@ attempt or owning phase:
|
||||
| `response.resolve`, `.authorize`, `.add_comment`, `.transaction*`, `.row_lock`, `.write`, `.publish` | Response selection, existing authorization, comment transaction including commit or rollback, row wait, insert, and external publication enqueue. |
|
||||
| `execution.realtime` | Synchronous live-event and plugin publication dispatch. It does not measure browser receipt or paint. |
|
||||
|
||||
For bounded native OpenCode diagnosis, set
|
||||
`PAPERCLIP_OPENCODE_PERFORMANCE_DIAGNOSTICS=1` as an agent/environment variable.
|
||||
The existing configured-environment projection carries it into the native Runner.
|
||||
It defaults off. The Runner creates an authenticated loopback OTLP sink and enables
|
||||
the pinned OpenCode runtime's internal tracing. The sink discards logs, resource
|
||||
attributes, span attributes, events, and unknown span names. It retains only closed
|
||||
phase names, wall timestamps, and success/failure, with a 1 MiB request limit,
|
||||
128 retained measurements, and 16 measurements per phase. No operator OTLP endpoint
|
||||
or ingestion credential is passed to OpenCode.
|
||||
|
||||
Available phases cover context/instruction loading, skill discovery, tool and model
|
||||
resolution, model streaming (`ai.streamText`, including stream consumption), Git
|
||||
discovery/default branch, snapshots, dependency installation, and plugin startup.
|
||||
These intervals overlap; do not sum them as elapsed time. At each terminal turn,
|
||||
already-received measurements travel in one `harness.diagnostic` event through the
|
||||
existing durable PRP channel. Late exports may be absent; collection never waits
|
||||
for a provider flush or delays terminal publication to complete the diagnostic.
|
||||
Late spans from a prior completed turn are discarded at the next boundary.
|
||||
|
||||
The server validates the closed fields again and records them under a foldable
|
||||
`provider.runtime` group inside `native.session.execute`. The normal operator
|
||||
endpoint gate controls Honeycomb export. The single diagnostic batch stays in the instance run log. Its OTel projection
|
||||
does not write duplicate phase rows or await additional database work before
|
||||
terminal consumption. Without an endpoint, no provider child spans are exported. Raw OpenCode OTLP must
|
||||
never be sent directly to the operator endpoint: it can contain prompts and
|
||||
authorization attributes before the loopback sink removes them.
|
||||
|
||||
Detail phases emit at most 16 spans of each name per attempt. Larger receipt and
|
||||
asset inventories continue under the containing group without unbounded per-row
|
||||
or per-file spans. Adding detail spans does not itself change work, query order,
|
||||
@@ -1954,6 +1991,12 @@ It carries no skill contents, identifiers, locations, or credentials. It uses th
|
||||
existing run performance events and operator-configured OpenTelemetry endpoint;
|
||||
no first-party Telemetry event is added.
|
||||
|
||||
The `skills.*` detail spans above distinguish inventory reconciliation and lock
|
||||
acquisition from repository transfer and file validation. A repository-backed
|
||||
skill's authored `SKILL.md` is not its complete runtime payload: the generated
|
||||
discovery entrypoint includes the pinned repository under `.paperclip-repository`.
|
||||
Measure the delivered bundle's file count and bytes when reproducing its cost.
|
||||
|
||||
Runtime preparation refreshes the company inventory once per listing. Local and
|
||||
catalog directories remain direct sources, so edits are visible on the next
|
||||
preparation. Explicit version selections still use their stored snapshots.
|
||||
|
||||
@@ -43,6 +43,7 @@ const OPENCODE_PROVIDER_ENVIRONMENT_KEYS: &[&str] = &[
|
||||
"PAPERCLIP_NATIVE_MCP_TOKEN",
|
||||
"PAPERCLIP_OPENCODE_PERMISSION_MODE",
|
||||
"PAPERCLIP_OPENCODE_RUNTIME_DIR",
|
||||
"PAPERCLIP_OPENCODE_PERFORMANCE_DIAGNOSTICS",
|
||||
"PAPERCLIP_RUNNER_INSTANCE_ID",
|
||||
"PAPERCLIP_RUN_ID",
|
||||
"PAPERCLIP_NORMALIZED_SESSION_ID",
|
||||
|
||||
@@ -671,6 +671,27 @@ fn measurement(value: &Value) -> Value {
|
||||
|
||||
/// Converts Codex app-server notifications into provider-neutral PRP events.
|
||||
/// Provider-native envelopes are consumed here and never cross the PRP boundary.
|
||||
fn provider_performance_spans(params: &Value) -> Vec<Value> {
|
||||
let mut counts = std::collections::BTreeMap::<&str, usize>::new();
|
||||
params.get("spans").and_then(Value::as_array).into_iter().flatten().take(128).filter_map(|span| {
|
||||
let name = span.get("name")?.as_str()?;
|
||||
if !matches!(name,
|
||||
"provider.context.environment" | "provider.context.instructions" | "provider.skills.discovery"
|
||||
| "provider.tools.resolve" | "provider.model.resolve" | "provider.model.stream"
|
||||
| "provider.workspace.snapshot.track" | "provider.workspace.snapshot.patch" | "provider.workspace.snapshot.diff"
|
||||
| "provider.workspace.git.discover" | "provider.workspace.git.default_branch"
|
||||
| "provider.dependencies.install" | "provider.plugins.initialize") { return None; }
|
||||
let start = span.get("startedAtMs")?.as_f64()?;
|
||||
let end = span.get("endedAtMs")?.as_f64()?;
|
||||
if !start.is_finite() || !end.is_finite() || start < 0.0 || end < start || end - start > 86_400_000.0 { return None; }
|
||||
let count = counts.entry(name).or_default();
|
||||
if *count >= 16 { return None; }
|
||||
*count += 1;
|
||||
Some(json!({"name":name,"startedAtMs":start,"endedAtMs":end,
|
||||
"outcome":if span.get("outcome").and_then(Value::as_str)==Some("failed") {"failed"} else {"ok"}}))
|
||||
}).collect()
|
||||
}
|
||||
|
||||
pub fn normalize_codex_notification(method: &str, params: &Value) -> Vec<NormalizedProviderEvent> {
|
||||
let mut events = Vec::new();
|
||||
let push = |events: &mut Vec<NormalizedProviderEvent>,
|
||||
@@ -685,6 +706,17 @@ pub fn normalize_codex_notification(method: &str, params: &Value) -> Vec<Normali
|
||||
};
|
||||
|
||||
match method {
|
||||
"paperclip/providerPerformance" => {
|
||||
let spans = provider_performance_spans(params);
|
||||
if !spans.is_empty() {
|
||||
push(
|
||||
&mut events,
|
||||
"harness.diagnostic",
|
||||
EventPriority::P1,
|
||||
json!({"code":"provider_performance_spans","spans":spans}),
|
||||
);
|
||||
}
|
||||
}
|
||||
"thread/compacted" => push(
|
||||
&mut events,
|
||||
"context.compacted",
|
||||
@@ -1464,6 +1496,37 @@ fn has_rfc_uri_scheme_prefix(value: &str) -> bool {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn performance_diagnostics_keep_only_bounded_content_free_measurements() {
|
||||
let mut spans = vec![json!({"name":"private prompt","startedAtMs":1,"endedAtMs":2})];
|
||||
for _ in 0..200 {
|
||||
spans.push(
|
||||
json!({"name":"provider.model.stream","startedAtMs":1000,"endedAtMs":1010,
|
||||
"outcome":"failed","authorization":"private","attributes":{"prompt":"private"}}),
|
||||
);
|
||||
}
|
||||
let events =
|
||||
normalize_codex_notification("paperclip/providerPerformance", &json!({"spans":spans}));
|
||||
assert_eq!(events.len(), 1);
|
||||
assert_eq!(events[0].event_type, "harness.diagnostic");
|
||||
assert_eq!(events[0].priority, EventPriority::P1);
|
||||
let rows = events[0].payload["spans"].as_array().unwrap();
|
||||
assert_eq!(rows.len(), 16);
|
||||
assert_eq!(
|
||||
rows[0],
|
||||
json!({"name":"provider.model.stream","startedAtMs":1000.0,"endedAtMs":1010.0,"outcome":"failed"})
|
||||
);
|
||||
assert!(!events[0].payload.to_string().contains("private"));
|
||||
assert!(normalize_codex_notification(
|
||||
"paperclip/providerPerformance",
|
||||
&json!({"spans":[
|
||||
{"name":"provider.model.stream","startedAtMs":10,"endedAtMs":1},
|
||||
{"name":"provider.model.stream","startedAtMs":0,"endedAtMs":86400001}
|
||||
]})
|
||||
)
|
||||
.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unrelated_information_retains_only_bounded_run_log_diagnostics() {
|
||||
let events = normalize_codex_notification(
|
||||
|
||||
@@ -30,6 +30,7 @@ import {
|
||||
withoutAmbientOpenCodeCommand,
|
||||
} from "./opencode-proxy-command.js";
|
||||
import { OpenCodeProxyUsageLedger } from "./opencode-proxy-usage.js";
|
||||
import { parseProviderPerformanceSpans, PROVIDER_PERFORMANCE_DIAGNOSTIC } from "../contracts/provider-performance.js";
|
||||
|
||||
type RpcMessage = {
|
||||
id?: string | number;
|
||||
@@ -307,6 +308,13 @@ async function pumpEvents(opened: HarnessSession): Promise<void> {
|
||||
);
|
||||
activeTurnId = null;
|
||||
} else if (event.eventType === "harness.diagnostic") {
|
||||
if (payload.code === PROVIDER_PERFORMANCE_DIAGNOSTIC) {
|
||||
send({method: "paperclip/providerPerformance", params: {
|
||||
threadId: opened.ids().driverSessionId, turnId: event.turnId,
|
||||
spans: parseProviderPerformanceSpans(payload, Date.now()),
|
||||
}});
|
||||
continue;
|
||||
}
|
||||
send({
|
||||
method: "warning",
|
||||
params: {
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
/** Content-free provider diagnostics carried by the existing PRP diagnostic event. */
|
||||
export const PROVIDER_PERFORMANCE_DIAGNOSTIC = "provider_performance_spans";
|
||||
export const PROVIDER_PERFORMANCE_ENV = "PAPERCLIP_OPENCODE_PERFORMANCE_DIAGNOSTICS";
|
||||
export const MAX_PROVIDER_PERFORMANCE_SPANS = 128;
|
||||
export const PROVIDER_PERFORMANCE_PHASES = [
|
||||
"provider.context.environment", "provider.context.instructions", "provider.skills.discovery",
|
||||
"provider.tools.resolve", "provider.model.resolve", "provider.model.stream",
|
||||
"provider.workspace.snapshot.track", "provider.workspace.snapshot.patch", "provider.workspace.snapshot.diff",
|
||||
"provider.workspace.git.discover", "provider.workspace.git.default_branch",
|
||||
"provider.dependencies.install", "provider.plugins.initialize",
|
||||
] as const;
|
||||
export type ProviderPerformancePhase = typeof PROVIDER_PERFORMANCE_PHASES[number];
|
||||
export interface ProviderPerformanceSpan {
|
||||
name: ProviderPerformancePhase;
|
||||
startedAtMs: number;
|
||||
endedAtMs: number;
|
||||
outcome: "ok" | "failed";
|
||||
}
|
||||
|
||||
const names = new Set<string>(PROVIDER_PERFORMANCE_PHASES);
|
||||
const maxDurationMs = 24 * 60 * 60 * 1000;
|
||||
|
||||
/** Project fields rather than forwarding provider attributes or untrusted diagnostic data. */
|
||||
export function parseProviderPerformanceSpans(payload: unknown, observedAtMs: number): ProviderPerformanceSpan[] {
|
||||
if (!payload || typeof payload !== "object" || Array.isArray(payload)) return [];
|
||||
const diagnostic = payload as Record<string, unknown>;
|
||||
if (diagnostic.code !== PROVIDER_PERFORMANCE_DIAGNOSTIC || !Array.isArray(diagnostic.spans)) return [];
|
||||
const counts = new Map<string, number>();
|
||||
return diagnostic.spans.slice(0, MAX_PROVIDER_PERFORMANCE_SPANS).flatMap((value): ProviderPerformanceSpan[] => {
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) return [];
|
||||
const row = value as Record<string, unknown>;
|
||||
const { name, startedAtMs, endedAtMs } = row;
|
||||
if (typeof name !== "string" || !names.has(name) || typeof startedAtMs !== "number" || typeof endedAtMs !== "number"
|
||||
|| !Number.isFinite(startedAtMs) || !Number.isFinite(endedAtMs)
|
||||
|| startedAtMs < observedAtMs - maxDurationMs || endedAtMs > observedAtMs + 1000
|
||||
|| endedAtMs < startedAtMs || endedAtMs - startedAtMs > maxDurationMs) return [];
|
||||
const count = counts.get(name) ?? 0;
|
||||
if (count >= 16) return [];
|
||||
counts.set(name, count + 1);
|
||||
return [{ name: name as ProviderPerformancePhase, startedAtMs, endedAtMs,
|
||||
outcome: row.outcome === "failed" ? "failed" : "ok" }];
|
||||
});
|
||||
}
|
||||
@@ -3503,6 +3503,7 @@ const runnerExplicitProviderEnvironmentKeys = [
|
||||
"AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE",
|
||||
"PAPERCLIP_OPENCODE_PERMISSION_MODE",
|
||||
"PAPERCLIP_OPENCODE_RUNTIME_DIR",
|
||||
"PAPERCLIP_OPENCODE_PERFORMANCE_DIAGNOSTICS",
|
||||
"PAPERCLIP_RUNNER_INSTANCE_ID",
|
||||
"PAPERCLIP_RUN_ID",
|
||||
"PAPERCLIP_NORMALIZED_SESSION_ID",
|
||||
|
||||
@@ -24,6 +24,8 @@ import { pipeline } from "node:stream/promises";
|
||||
import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path";
|
||||
|
||||
import { parseOpenCodeReasoningMode, type OpenCodeReasoningMode } from "./reasoning-mode.js";
|
||||
import { startOpenCodePerformanceDiagnostics } from "./performance-diagnostics.js";
|
||||
import { PROVIDER_PERFORMANCE_ENV } from "../../contracts/provider-performance.js";
|
||||
import { annotateOmittedProviderText, parseOpenCodeSseFrames } from "./sse-frames.js";
|
||||
import { githubCredentialEnvironment } from "../../github-credential-environment.js";
|
||||
|
||||
@@ -217,6 +219,7 @@ interface OpenCodeRuntime {
|
||||
bridge: OpenCodeMcpBridge;
|
||||
trace: ProviderTraceFileSink | null;
|
||||
sensitiveValues: readonly string[];
|
||||
performanceDiagnostics: Awaited<ReturnType<typeof startOpenCodePerformanceDiagnostics>> | null;
|
||||
close(input?: {
|
||||
finalizeTrace?: boolean;
|
||||
reason?: string | null;
|
||||
@@ -520,6 +523,7 @@ class OpenCodeHarnessSession implements HarnessSession {
|
||||
#activeTraceEmittedEventIds: string[] = [];
|
||||
#sourceSequence: number;
|
||||
#activeTurnId: string | null;
|
||||
#performanceBoundaryMs = 0;
|
||||
#result: PrpStructuredRunResult | null;
|
||||
#resultFingerprint: string | null;
|
||||
#resultCallId: string | null;
|
||||
@@ -2133,6 +2137,11 @@ class OpenCodeHarnessSession implements HarnessSession {
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (["turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"].includes(eventType)) {
|
||||
const performance = this.#runtime.performanceDiagnostics?.take(this.#performanceBoundaryMs);
|
||||
this.#performanceBoundaryMs = this.#now().getTime();
|
||||
if (performance) this.#emit("harness.diagnostic", performance, refs);
|
||||
}
|
||||
const sourceSeq = ++this.#sourceSequence;
|
||||
const event: PrpEvent = {
|
||||
schema: "paperclip.prp.event.v1",
|
||||
@@ -2349,6 +2358,15 @@ async function startRuntime(input: {
|
||||
input.trace?.addSensitiveValues([providerProxy.token]);
|
||||
}
|
||||
let child: ChildProcess | undefined;
|
||||
// Failure to collect diagnostics must not prevent the requested work.
|
||||
const performanceDiagnostics = input.options.environment?.[PROVIDER_PERFORMANCE_ENV] === "1"
|
||||
? await startOpenCodePerformanceDiagnostics().catch(() => null) : null;
|
||||
if (performanceDiagnostics) {
|
||||
const values = [performanceDiagnostics.environment.OTEL_EXPORTER_OTLP_HEADERS,
|
||||
performanceDiagnostics.environment.OTEL_EXPORTER_OTLP_HEADERS.replace("Authorization=Bearer ", "")];
|
||||
sensitiveValues.push(...values);
|
||||
input.trace?.addSensitiveValues(values);
|
||||
}
|
||||
let launchPrepared = false;
|
||||
const releaseExecutable = () => {
|
||||
if (!launchPrepared) return;
|
||||
@@ -2366,6 +2384,7 @@ async function startRuntime(input: {
|
||||
small_model: input.options.model,
|
||||
share: "disabled",
|
||||
autoupdate: false,
|
||||
...(performanceDiagnostics ? { experimental: { openTelemetry: true } } : {}),
|
||||
// The configured entry is already composed exactly once into the session
|
||||
// system prompt; siblings remain available through the read-only root.
|
||||
instructions: [],
|
||||
@@ -2450,6 +2469,7 @@ async function startRuntime(input: {
|
||||
OPENCODE_DISABLE_DEFAULT_PLUGINS: "true",
|
||||
OPENCODE_SERVER_USERNAME: username,
|
||||
OPENCODE_SERVER_PASSWORD: password,
|
||||
...performanceDiagnostics?.environment,
|
||||
...(providerProxy ? { NO_PROXY: [input.options.environment?.no_proxy ?? input.options.environment?.NO_PROXY, "127.0.0.1", "localhost"].filter(Boolean).join(",") } : {}),
|
||||
},
|
||||
);
|
||||
@@ -2547,6 +2567,7 @@ async function startRuntime(input: {
|
||||
bridge,
|
||||
trace: input.trace,
|
||||
sensitiveValues,
|
||||
performanceDiagnostics,
|
||||
close: async (closeInput = {}) => {
|
||||
await providerProxy?.close();
|
||||
await bridge.close().catch(() => {});
|
||||
@@ -2570,6 +2591,7 @@ async function startRuntime(input: {
|
||||
}
|
||||
await waitForExit(providerChild, 2_000);
|
||||
}
|
||||
await performanceDiagnostics?.close().catch(() => {});
|
||||
await rm(join(configHome, "opencode", "opencode.json"), {
|
||||
force: true,
|
||||
}).catch(() => undefined);
|
||||
@@ -2580,6 +2602,7 @@ async function startRuntime(input: {
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
await performanceDiagnostics?.close().catch(() => {});
|
||||
await providerProxy?.close();
|
||||
await bridge.close().catch(() => {});
|
||||
child?.kill("SIGKILL");
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { parseProviderPerformanceSpans, PROVIDER_PERFORMANCE_DIAGNOSTIC } from "../../contracts/provider-performance.js";
|
||||
import { startOpenCodePerformanceDiagnostics } from "./performance-diagnostics.js";
|
||||
|
||||
function batch(spans: unknown[]) { return { resourceSpans: [{resource:{attributes:[{key:"secret",value:"credential"}]},scopeSpans:[{spans}]}]}; }
|
||||
function span(name = "ai.streamText", extra = {}) {
|
||||
const now=Date.now();
|
||||
return {name,startTimeUnixNano:String(BigInt(now-10)*1000000n),endTimeUnixNano:String(BigInt(now)*1000000n),
|
||||
attributes:[{key:"Authorization",value:{stringValue:"Bearer private"}},{key:"prompt",value:{stringValue:"private task"}}],
|
||||
events:[{name:"private content"}],...extra};
|
||||
}
|
||||
|
||||
describe("OpenCode content-free performance sink",()=>{
|
||||
it("accepts authenticated loopback OTLP and projects only closed phase names and timings",async()=>{
|
||||
const sink=await startOpenCodePerformanceDiagnostics();
|
||||
const url=sink.environment.OTEL_EXPORTER_OTLP_ENDPOINT;
|
||||
const headers={Authorization:sink.environment.OTEL_EXPORTER_OTLP_HEADERS.slice("Authorization=".length)};
|
||||
try {
|
||||
expect((await fetch(url+"/v1/traces",{method:"POST",body:JSON.stringify(batch([span()]))})).status).toBe(401);
|
||||
expect(sink.take()).toBeNull();
|
||||
expect((await fetch(url+"/v1/traces",{method:"POST",headers,body:JSON.stringify(batch([
|
||||
span(),span("Snapshot.track",{status:{code:2}}),span("private-provider-name"),span("__proto__"),span("toString"),
|
||||
]))})).status).toBe(200);
|
||||
const diagnostic=sink.take()!;
|
||||
expect(diagnostic.code).toBe(PROVIDER_PERFORMANCE_DIAGNOSTIC);
|
||||
expect(diagnostic.spans.map(s=>[s.name,s.outcome])).toEqual([["provider.model.stream","ok"],["provider.workspace.snapshot.track","failed"]]);
|
||||
expect(diagnostic.spans[0].endedAtMs-diagnostic.spans[0].startedAtMs).toBeCloseTo(10, 2);
|
||||
expect(JSON.stringify(diagnostic)).not.toMatch(/private|Authorization|prompt|credential/);
|
||||
expect(Object.keys(diagnostic.spans[0]).sort()).toEqual(["endedAtMs","name","outcome","startedAtMs"]);
|
||||
expect(sink.take()).toBeNull();
|
||||
expect((await fetch(url+"/v1/logs",{method:"POST",headers,body:'{"private":"discard"}'})).status).toBe(200);
|
||||
expect(sink.take()).toBeNull();
|
||||
} finally {await sink.close();}
|
||||
await expect(fetch(url+"/v1/traces")).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("bounds bodies, malformed timestamps and repeated spans and drops prior-turn exports",async()=>{
|
||||
const sink=await startOpenCodePerformanceDiagnostics();
|
||||
const url=sink.environment.OTEL_EXPORTER_OTLP_ENDPOINT;
|
||||
const headers={Authorization:sink.environment.OTEL_EXPORTER_OTLP_HEADERS.slice("Authorization=".length)};
|
||||
const post=(body:string)=>fetch(url+"/v1/traces",{method:"POST",headers,body});
|
||||
try {
|
||||
expect((await post("x".repeat(1024*1024+1))).status).toBe(413);
|
||||
expect((await post("invalid json")).status).toBe(400);
|
||||
expect((await post(JSON.stringify(batch([
|
||||
span("ai.streamText",{endTimeUnixNano:"NaN"}),span("ai.streamText",{endTimeUnixNano:"1"}),
|
||||
...Array.from({length:200},()=>span()),
|
||||
])))).status).toBe(200);
|
||||
expect(sink.take()!.spans).toHaveLength(16);
|
||||
await post(JSON.stringify(batch([span()])));
|
||||
expect(sink.take(Date.now()+100)).toBeNull();
|
||||
}finally{await sink.close();}
|
||||
});
|
||||
|
||||
it("validates the shared diagnostic again before server export",()=>{
|
||||
const now=Date.now();
|
||||
const safe={name:"provider.model.stream",startedAtMs:now-20,endedAtMs:now,outcome:"ok",prompt:"private"};
|
||||
const parsed=parseProviderPerformanceSpans({code:PROVIDER_PERFORMANCE_DIAGNOSTIC,spans:[
|
||||
safe,{...safe,name:"private"},{...safe,startedAtMs:Infinity},{...safe,endedAtMs:now+2000},
|
||||
{...safe,startedAtMs:now-86400001},{...safe,endedAtMs:now-30},
|
||||
]},now);
|
||||
expect(parsed).toEqual([{name:safe.name,startedAtMs:now-20,endedAtMs:now,outcome:"ok"}]);
|
||||
expect(parseProviderPerformanceSpans({code:"other",spans:[safe]},now)).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,103 @@
|
||||
import { randomBytes, timingSafeEqual } from "node:crypto";
|
||||
import { createServer } from "node:http";
|
||||
import {
|
||||
MAX_PROVIDER_PERFORMANCE_SPANS, PROVIDER_PERFORMANCE_DIAGNOSTIC,
|
||||
type ProviderPerformancePhase, type ProviderPerformanceSpan,
|
||||
} from "../../contracts/provider-performance.js";
|
||||
|
||||
const phases: Readonly<Record<string, ProviderPerformancePhase>> = {
|
||||
"SystemPrompt.environment": "provider.context.environment",
|
||||
"Instruction.system": "provider.context.instructions",
|
||||
"Skill.discovery": "provider.skills.discovery",
|
||||
"SessionTools.resolve": "provider.tools.resolve",
|
||||
"Provider.getModel": "provider.model.resolve",
|
||||
"ai.streamText": "provider.model.stream",
|
||||
"Snapshot.track": "provider.workspace.snapshot.track",
|
||||
"Snapshot.patch": "provider.workspace.snapshot.patch",
|
||||
"Snapshot.diffFull": "provider.workspace.snapshot.diff",
|
||||
"Git.repo.discover": "provider.workspace.git.discover",
|
||||
"Git.defaultBranch": "provider.workspace.git.default_branch",
|
||||
"Npm.install": "provider.dependencies.install",
|
||||
"PluginInternal.boot": "provider.plugins.initialize",
|
||||
};
|
||||
const maxBodyBytes = 1024 * 1024;
|
||||
|
||||
function object(value: unknown): Record<string, unknown> {
|
||||
return value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
}
|
||||
|
||||
/** OpenCode's raw OTLP can contain authorization and prompt attributes. None leave this sink. */
|
||||
export async function startOpenCodePerformanceDiagnostics() {
|
||||
const token = randomBytes(32).toString("hex");
|
||||
const authorization = Buffer.from(`Bearer ${token}`);
|
||||
const spans: ProviderPerformanceSpan[] = [];
|
||||
const counts = new Map<string, number>();
|
||||
const server = createServer((request, response) => {
|
||||
const provided = Buffer.from(request.headers.authorization ?? "");
|
||||
if (provided.length !== authorization.length || !timingSafeEqual(provided, authorization)) {
|
||||
request.resume(); response.writeHead(401).end(); return;
|
||||
}
|
||||
if (request.method !== "POST" || !["/v1/traces", "/v1/logs"].includes(request.url ?? "")) {
|
||||
request.resume(); response.writeHead(404).end(); return;
|
||||
}
|
||||
let bytes = 0;
|
||||
const chunks: Buffer[] = [];
|
||||
request.on("data", (chunk: Buffer) => {
|
||||
bytes += chunk.length;
|
||||
if (bytes <= maxBodyBytes && request.url === "/v1/traces") chunks.push(chunk);
|
||||
else chunks.length = 0;
|
||||
});
|
||||
request.on("error", () => response.destroy());
|
||||
request.on("end", () => {
|
||||
if (bytes > maxBodyBytes) { response.writeHead(413).end(); return; }
|
||||
if (request.url === "/v1/logs") { response.writeHead(200).end("{}"); return; }
|
||||
try {
|
||||
const batch = object(JSON.parse(Buffer.concat(chunks).toString("utf8")));
|
||||
for (const resource of Array.isArray(batch.resourceSpans) ? batch.resourceSpans : []) {
|
||||
for (const scope of Array.isArray(object(resource).scopeSpans) ? object(resource).scopeSpans as unknown[] : []) {
|
||||
for (const value of Array.isArray(object(scope).spans) ? object(scope).spans as unknown[] : []) {
|
||||
const row = object(value);
|
||||
const name = typeof row.name === "string" && Object.hasOwn(phases, row.name) ? phases[row.name] : undefined;
|
||||
if (!name || spans.length >= MAX_PROVIDER_PERFORMANCE_SPANS || (counts.get(name) ?? 0) >= 16) continue;
|
||||
const startedAtMs = Number(row.startTimeUnixNano) / 1e6;
|
||||
const endedAtMs = Number(row.endTimeUnixNano) / 1e6;
|
||||
const now = Date.now();
|
||||
if (!Number.isFinite(startedAtMs) || !Number.isFinite(endedAtMs) || startedAtMs < now - 86400000
|
||||
|| endedAtMs < startedAtMs || endedAtMs > now + 1000) continue;
|
||||
spans.push({name, startedAtMs, endedAtMs, outcome: object(row.status).code === 2 ? "failed" : "ok"});
|
||||
counts.set(name, (counts.get(name) ?? 0) + 1);
|
||||
}
|
||||
}
|
||||
}
|
||||
response.writeHead(200, {"Content-Type":"application/json"}).end("{}");
|
||||
} catch { response.writeHead(400).end(); }
|
||||
});
|
||||
});
|
||||
server.requestTimeout = 2000;
|
||||
server.headersTimeout = 2000;
|
||||
server.maxConnections = 2;
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(0, "127.0.0.1", () => { server.off("error", reject); resolve(); });
|
||||
});
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") { server.close(); throw new Error("OpenCode diagnostics did not bind loopback"); }
|
||||
return {
|
||||
environment: {
|
||||
OTEL_EXPORTER_OTLP_ENDPOINT: `http://127.0.0.1:${address.port}`,
|
||||
OTEL_EXPORTER_OTLP_PROTOCOL: "http/json",
|
||||
// OpenCode's Effect logger reads literal values; its trace exporter receives the same map.
|
||||
OTEL_EXPORTER_OTLP_HEADERS: `Authorization=Bearer ${token}`,
|
||||
OTEL_BSP_SCHEDULE_DELAY: "100",
|
||||
},
|
||||
take(sinceMs = 0): {code: string; spans: ProviderPerformanceSpan[]} | null {
|
||||
const collected = spans.splice(0).filter(span => span.startedAtMs >= sinceMs);
|
||||
counts.clear();
|
||||
return collected.length ? {code: PROVIDER_PERFORMANCE_DIAGNOSTIC, spans: collected} : null;
|
||||
},
|
||||
async close() {
|
||||
spans.length = 0;
|
||||
await new Promise<void>(resolve => { server.close(() => resolve()); server.closeAllConnections(); });
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -4,6 +4,7 @@ export { CURSOR_DISTRIBUTION_PINS } from "./drivers/acpx/generated-profiles.js";
|
||||
export * from "./contracts/control-plane-port.js";
|
||||
export * from "./contracts/completion-result.js";
|
||||
export * from "./contracts/codex.js";
|
||||
export * from "./contracts/provider-performance.js";
|
||||
export * from "./contracts/durable-recovery.js";
|
||||
export * from "./contracts/harness-driver.js";
|
||||
export * from "./contracts/local-runner.js";
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { bindRunnerLaunchResources, createRunnerProviderLaunch, runnerLaunchRequirementsDigest, type PreparedRunnerProviderLaunch, type RunnerProviderLaunchRequirements } from "./runner-provider-launch.js";
|
||||
import { coldAdmissionTimeoutMs, waitForRunnerCommand } from "./runnerd-codex-transport.js";
|
||||
import { configuredEnvironmentProjection } from "../configured-environment.js";
|
||||
import {
|
||||
chmod,
|
||||
cp,
|
||||
@@ -2086,6 +2087,9 @@ it("preserves OpenCode runtime bindings when a durable runner is respawned", ()
|
||||
environment: {
|
||||
PATH: "/bin",
|
||||
OPENROUTER_API_KEY: "test-provider-key",
|
||||
...configuredEnvironmentProjection({ PAPERCLIP_OPENCODE_PERFORMANCE_DIAGNOSTICS: "1" }),
|
||||
OTEL_EXPORTER_OTLP_ENDPOINT: "https://must-not-reach-opencode.invalid",
|
||||
OTEL_EXPORTER_OTLP_HEADERS: "Authorization=private",
|
||||
HOME: "/host/home",
|
||||
CODEX_HOME: "/host/codex-home",
|
||||
DATABASE_URL: "must-not-reach-runnerd",
|
||||
@@ -2115,7 +2119,10 @@ it("preserves OpenCode runtime bindings when a durable runner is respawned", ()
|
||||
PAPERCLIP_NORMALIZED_SESSION_ID: "session-1",
|
||||
PAPERCLIP_NATIVE_RUNTIME_CONTEXT_PATH: "/isolated/runtime-context.json",
|
||||
OPENROUTER_API_KEY: "test-provider-key",
|
||||
PAPERCLIP_OPENCODE_PERFORMANCE_DIAGNOSTICS: "1",
|
||||
});
|
||||
expect(environment.OTEL_EXPORTER_OTLP_ENDPOINT).toBeUndefined();
|
||||
expect(environment.OTEL_EXPORTER_OTLP_HEADERS).toBeUndefined();
|
||||
expect(environment.HOME).toBeUndefined();
|
||||
expect(environment.CODEX_HOME).toBeUndefined();
|
||||
expect(environment.DATABASE_URL).toBeUndefined();
|
||||
@@ -9066,7 +9073,9 @@ it.each([
|
||||
opencodeProxySha256: digest(proxy),
|
||||
providerNodeCommand: providerNode,
|
||||
providerNodeCommandSha256: digest(providerNode),
|
||||
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
|
||||
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key",
|
||||
...(supportsReasoning ? configuredEnvironmentProjection({ PAPERCLIP_OPENCODE_PERFORMANCE_DIAGNOSTICS: "1" }) : {}),
|
||||
},
|
||||
});
|
||||
const task = createCodexTaskEnvelope({
|
||||
objective: "Preserve the prepared task.", contractRevision: "prepared-v1",
|
||||
@@ -9113,6 +9122,14 @@ it.each([
|
||||
expect(events.find(event => event.eventType === "session.started")?.payload.context).toMatchObject({ model });
|
||||
expect(completionCalls).toBe(2);
|
||||
expect(events.filter(event => event.eventType === "run.result.proposed")).toHaveLength(1);
|
||||
const performance = events.filter(event => event.eventType === "harness.diagnostic" && event.payload.code === "provider_performance_spans");
|
||||
expect(performance).toHaveLength(supportsReasoning ? 1 : 0);
|
||||
if (supportsReasoning) {
|
||||
expect(performance[0].payload.spans).toEqual([{
|
||||
name: "provider.model.stream", startedAtMs: expect.any(Number), endedAtMs: expect.any(Number), outcome: "ok",
|
||||
}]);
|
||||
expect(JSON.stringify(performance)).not.toContain("private-performance-sentinel");
|
||||
}
|
||||
const sessionRoots = (await readdir(runtime, { withFileTypes: true })).filter((entry) => entry.isDirectory());
|
||||
expect(sessionRoots).toHaveLength(1);
|
||||
const requests = (await readFile(join(runtime, sessionRoots[0]!.name, "data/fake-prompt-requests.ndjson"), "utf8")).trim().split("\n").map((line) => JSON.parse(line));
|
||||
|
||||
@@ -128,6 +128,24 @@ let promptTurnSeq = 0;
|
||||
// sealed.
|
||||
let lateStragglerRemaining = 0;
|
||||
function emit(value) {
|
||||
if (value.type === "session.idle" && runtimeConfig.experimental?.openTelemetry) {
|
||||
const now=Date.now();
|
||||
const headers=Object.fromEntries((process.env.OTEL_EXPORTER_OTLP_HEADERS??"").split(",").map(entry=>{
|
||||
const i=entry.indexOf("=");return [entry.slice(0,i),entry.slice(i+1)];
|
||||
}));
|
||||
void fetch(`${process.env.OTEL_EXPORTER_OTLP_ENDPOINT}/v1/traces`,{method:"POST",headers,
|
||||
body:JSON.stringify({resourceSpans:[{scopeSpans:[{spans:[{name:"ai.streamText",
|
||||
startTimeUnixNano:String(BigInt(now-22)*1000000n),endTimeUnixNano:String(BigInt(now)*1000000n),
|
||||
attributes:[{key:"prompt",value:{stringValue:"private-performance-sentinel"}}],
|
||||
}]}]}]})}).then(response=>{
|
||||
if(!response.ok)throw new Error(`Performance fixture rejected: ${response.status}`);
|
||||
emitFrame(value);
|
||||
}).catch(error=>{process.stderr.write(String(error));process.exit(1)});
|
||||
return;
|
||||
}
|
||||
emitFrame(value);
|
||||
}
|
||||
function emitFrame(value) {
|
||||
const scoped =
|
||||
promptTurnSeq > 0 && typeof value.id === "string"
|
||||
? { ...value, id: `${value.id}#${promptTurnSeq}` }
|
||||
|
||||
@@ -0,0 +1,260 @@
|
||||
/** Real skill resolution, PostgreSQL, and filesystem work; no provider is started. */
|
||||
import assert from "node:assert/strict";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { promises as fs } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { connect, createServer, type Socket } from "node:net";
|
||||
import { context, trace } from "@opentelemetry/api";
|
||||
import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks";
|
||||
import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace";
|
||||
import { resourceFromAttributes } from "@opentelemetry/resources";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { authUsers, companyMemberships, companySkillSources, heartbeatRuns, projects, projectSkillSources,
|
||||
companies, companySkills, companySkillVersions, companySkillRepositorySnapshots,
|
||||
applyPendingMigrations, closeRegisteredClients, createDb, startEmbeddedPostgresTestDatabase } from "../../packages/db/src/index.js";
|
||||
import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared";
|
||||
import { createRunPreparationTiming, withRunExecutionTiming } from "../src/services/execution-preparation-timing.js";
|
||||
import { repositorySkillProjection } from "../src/services/skill-repository-package.js";
|
||||
import { skillFileBytes, skillSnapshotHash } from "../src/services/skill-snapshot.js";
|
||||
|
||||
const argument = (name: string) => process.argv.find(value => value.startsWith(`${name}=`))?.slice(name.length + 1);
|
||||
const exportOtlp = argument("--export-otlp") === "true";
|
||||
const authenticated = argument("--actor") === "agent";
|
||||
const stagingShape = argument("--staging-shape") === "true";
|
||||
// This standalone benchmark owns its tracer. Load the service graph with the
|
||||
// app-wide SDK disabled, then restore the operator's export configuration.
|
||||
const operatorEndpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
|
||||
delete process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
|
||||
let companySkillService: typeof import("../src/services/company-skills.js").companySkillService;
|
||||
let withSkillActor: typeof import("../src/services/skill-access.js").withSkillActor;
|
||||
let createAgentLifecycle: typeof import("../src/services/agent-lifecycle.js").createAgentLifecycle;
|
||||
try {
|
||||
({ companySkillService } = await import("../src/services/company-skills.js"));
|
||||
({ withSkillActor } = await import("../src/services/skill-access.js"));
|
||||
({ createAgentLifecycle } = await import("../src/services/agent-lifecycle.js"));
|
||||
}
|
||||
finally {
|
||||
if (operatorEndpoint !== undefined) process.env.OTEL_EXPORTER_OTLP_ENDPOINT = operatorEndpoint;
|
||||
}
|
||||
const sourceRevision = execFileSync("git", ["rev-parse", "HEAD"], {
|
||||
cwd: fileURLToPath(new URL("../../", import.meta.url)), encoding: "utf8",
|
||||
}).trim();
|
||||
const sourceHash = createHash("sha256");
|
||||
for (const file of [import.meta.url, new URL("../src/services/company-skills.ts", import.meta.url),
|
||||
new URL("../src/services/skill-version-cache.ts", import.meta.url), new URL("../src/services/execution-preparation-timing.ts", import.meta.url)]) {
|
||||
sourceHash.update(await fs.readFile(new URL(file)));
|
||||
}
|
||||
const sourceChangesHash = sourceHash.digest("hex");
|
||||
if (exportOtlp) assert.ok(process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim(), "OTLP export requires an operator-configured endpoint");
|
||||
const snapshotFile = argument("--snapshot-file");
|
||||
const entrypoint = argument("--entrypoint") ?? "skills/diagnostic/SKILL.md";
|
||||
const files: CompanySkillVersionFileInventoryEntry[] = snapshotFile
|
||||
? JSON.parse(await fs.readFile(snapshotFile, "utf8"))
|
||||
: [{ path: entrypoint, kind: "skill", content: "---\nname: diagnostic\ndescription: Native preparation benchmark\n---\nReport the model ID.\n" },
|
||||
...Array.from({ length: 207 }, (_, index) => ({ path: `assets/${index}.txt`, kind: "reference" as const, content: "x".repeat(52_000) }))];
|
||||
const projected = repositorySkillProjection(files, entrypoint);
|
||||
const previous = { home: process.env.PAPERCLIP_HOME, instance: process.env.PAPERCLIP_INSTANCE_ID, endpoint: process.env.OTEL_EXPORTER_OTLP_ENDPOINT };
|
||||
const home = await fs.mkdtemp(path.join(os.tmpdir(), "native-skill-benchmark-"));
|
||||
let provider: TracerProvider | undefined;
|
||||
let manager: AsyncLocalStorageContextManager | undefined;
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb> | undefined;
|
||||
let proxy: ReturnType<typeof createServer> | null = null;
|
||||
const connections = new Set<Socket>();
|
||||
try {
|
||||
process.env.PAPERCLIP_HOME = home;
|
||||
process.env.PAPERCLIP_INSTANCE_ID = "default";
|
||||
process.env.OTEL_EXPORTER_OTLP_ENDPOINT = exportOtlp ? previous.endpoint : "http://127.0.0.1:4318";
|
||||
const exporter = new InMemorySpanExporter();
|
||||
const processors = [new SimpleSpanProcessor({ exporter })];
|
||||
const exportResults: number[] = [];
|
||||
if (exportOtlp) {
|
||||
const packages: Record<string, string> = { grpc: "@opentelemetry/exporter-trace-otlp-grpc",
|
||||
"http/protobuf": "@opentelemetry/exporter-trace-otlp-proto", "http/json": "@opentelemetry/exporter-trace-otlp-http" };
|
||||
const packageName = packages[process.env.OTEL_EXPORTER_OTLP_PROTOCOL ?? "grpc"];
|
||||
assert.ok(packageName, "Unsupported OTLP protocol");
|
||||
const { OTLPTraceExporter } = await import(packageName);
|
||||
const live = new OTLPTraceExporter();
|
||||
processors.push(new SimpleSpanProcessor({ exporter: {
|
||||
export: (spans, callback) => live.export(spans, (result: { code: number }) => { exportResults.push(result.code); callback(result); }),
|
||||
shutdown: () => live.shutdown(),
|
||||
} }));
|
||||
}
|
||||
provider = new TracerProvider({ spanProcessors: processors, resource: resourceFromAttributes({
|
||||
"service.name": process.env.OTEL_SERVICE_NAME || "paperclip",
|
||||
"service.version": `local-skill-bench-${sourceChangesHash.slice(0, 12)}`,
|
||||
"deployment.environment.name": "local-diagnostic",
|
||||
}) });
|
||||
manager = new AsyncLocalStorageContextManager().enable();
|
||||
assert.ok(context.setGlobalContextManager(manager));
|
||||
assert.ok(trace.setGlobalTracerProvider(provider));
|
||||
async function startDatabase() {
|
||||
const port = argument("--postgres-port");
|
||||
if (!port) return startEmbeddedPostgresTestDatabase("native-skill-benchmark-");
|
||||
assert.match(port, /^\d{1,5}$/);
|
||||
assert.ok(Number(port) > 0 && Number(port) <= 65535);
|
||||
const username = argument("--postgres-user") ?? os.userInfo().username;
|
||||
const adminUrl = `postgres://${encodeURIComponent(username)}@127.0.0.1:${port}/postgres`;
|
||||
const name = `native_skill_bench_${randomUUID().replaceAll("-", "")}`;
|
||||
const connectionString = `postgres://${encodeURIComponent(username)}@127.0.0.1:${port}/${name}`;
|
||||
const admin = createDb(adminUrl, { maxConnections: 1 }).$client;
|
||||
try { await admin.unsafe(`CREATE DATABASE "${name}"`); }
|
||||
finally { await admin.end({ timeout: 1 }); }
|
||||
const cleanup = async () => {
|
||||
await closeRegisteredClients(connectionString);
|
||||
const cleanupAdmin = createDb(adminUrl, { maxConnections: 1 }).$client;
|
||||
try { await cleanupAdmin.unsafe(`DROP DATABASE "${name}"`); }
|
||||
finally { await cleanupAdmin.end({ timeout: 1 }); }
|
||||
};
|
||||
try { await applyPendingMigrations(connectionString); }
|
||||
catch (error) { await cleanup(); throw error; }
|
||||
return { connectionString, cleanup };
|
||||
}
|
||||
database = await startDatabase();
|
||||
const networkDelayMs = Number(argument("--network-delay-ms") ?? 0);
|
||||
assert.ok(Number.isFinite(networkDelayMs) && networkDelayMs >= 0 && networkDelayMs <= 100);
|
||||
const target = new URL(database.connectionString);
|
||||
const upstreamPort = Number(target.port);
|
||||
proxy = networkDelayMs ? createServer(client => {
|
||||
const upstream = connect({ host: "127.0.0.1", port: upstreamPort });
|
||||
for (const socket of [client, upstream]) {
|
||||
socket.setNoDelay(true);
|
||||
connections.add(socket);
|
||||
socket.on("error", () => { client.destroy(); upstream.destroy(); });
|
||||
socket.on("close", () => { connections.delete(socket); });
|
||||
}
|
||||
const relay = (source: Socket, destination: Socket) => {
|
||||
let pendingBytes = 0;
|
||||
source.on("data", chunk => {
|
||||
pendingBytes += chunk.length;
|
||||
if (pendingBytes >= 2 * 1024 * 1024) source.pause();
|
||||
setTimeout(() => {
|
||||
pendingBytes -= chunk.length;
|
||||
if (destination.destroyed) return;
|
||||
const writable = destination.write(chunk);
|
||||
if (writable && pendingBytes < 2 * 1024 * 1024) source.resume();
|
||||
}, networkDelayMs);
|
||||
});
|
||||
destination.on("drain", () => { if (pendingBytes < 2 * 1024 * 1024) source.resume(); });
|
||||
source.on("end", () => setTimeout(() => destination.end(), networkDelayMs));
|
||||
};
|
||||
relay(client, upstream); relay(upstream, client);
|
||||
}) : null;
|
||||
if (proxy) {
|
||||
const server = proxy;
|
||||
await new Promise<void>((resolve, reject) => { server.once("error", reject); server.listen(0, "127.0.0.1", resolve); });
|
||||
const address = server.address();
|
||||
assert.ok(address && typeof address !== "string");
|
||||
target.port = String(address.port);
|
||||
}
|
||||
db = createDb(target.href);
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const userId = randomUUID();
|
||||
const skillId = randomUUID();
|
||||
const key = `company/${companyId}/diagnostic`;
|
||||
const samples: unknown[] = [];
|
||||
let rootPoolQueries = 0;
|
||||
db.$client.options.debug = () => { rootPoolQueries++; };
|
||||
await db.insert(companies).values({ id: companyId, name: "Native preparation benchmark", issuePrefix: "BENCH" });
|
||||
if (authenticated) {
|
||||
await createAgentLifecycle(db).requestHire(companyId, {
|
||||
id: agentId, name: "Diagnostic reader", adapterType: "paperclip_runner", status: "paused",
|
||||
}, { runnerResolved: true });
|
||||
await db.insert(authUsers).values({ id: userId, name: "Diagnostic owner", email: `${userId}@example.invalid`, createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companyMemberships).values([
|
||||
{ companyId, principalType: "agent", principalId: agentId, membershipRole: "member" },
|
||||
{ companyId, principalType: "user", principalId: userId, membershipRole: "owner" },
|
||||
]);
|
||||
}
|
||||
if (stagingShape) {
|
||||
// Match observed staging cardinalities without copying unrelated company content.
|
||||
const projectRows = await db.insert(projects).values(Array.from({ length: 14 }, (_, i) => ({ companyId, name: `Fixture ${i}` }))).returning();
|
||||
const sourceRows = await db.insert(companySkillSources).values(Array.from({ length: 10 }, (_, i) => ({
|
||||
companyId, repositoryUrl: `https://example.invalid/fixture-${i}`, fullName: `fixture/${i}`, trackingRef: "main",
|
||||
accessScope: i < 4 ? "projects" as const : "company" as const,
|
||||
}))).returning();
|
||||
await db.insert(projectSkillSources).values(Array.from({ length: 8 }, (_, i) => ({ companyId,
|
||||
projectId: projectRows[i % 5]!.id, sourceId: sourceRows[i]!.id,
|
||||
repositoryUrl: sourceRows[i]!.repositoryUrl, trackingRef: "main", status: "ready" as const,
|
||||
})));
|
||||
await db.insert(companySkills).values(Array.from({ length: 155 }, (_, i) => ({ companyId,
|
||||
key: `fixture/unselected-${i}`, slug: `unselected-${i}`, name: `Unselected ${i}`, sourceType: "github",
|
||||
sourceLocator: "https://example.invalid/immutable-fixture", markdown: "x".repeat(7_330),
|
||||
accessSourceId: sourceRows[i % sourceRows.length]!.id,
|
||||
})));
|
||||
await db.insert(companySkills).values({ companyId, key: "fixture/catalog", slug: "catalog", name: "Unselected catalog", sourceType: "catalog", markdown: "x".repeat(1_578) });
|
||||
}
|
||||
await db.insert(companySkills).values({ id: skillId, companyId, key, slug: "diagnostic", name: "Diagnostic",
|
||||
markdown: files.find(file => file.path === entrypoint)!.content, sourceType: "github",
|
||||
sourceLocator: "https://example.invalid/immutable-fixture", metadata: { packageMode: "repository", snapshotHash: "fixture" } });
|
||||
const [snapshot] = await db.insert(companySkillRepositorySnapshots).values({ companyId, contentHash: skillSnapshotHash(files), fileInventory: files }).returning();
|
||||
const [version] = await db.insert(companySkillVersions).values({ companyId, companySkillId: skillId, revisionNumber: 1,
|
||||
repositorySnapshotId: snapshot!.id, repositorySkillPath: entrypoint }).returning();
|
||||
await db.update(companySkills).set({ currentVersionId: version!.id }).where(eq(companySkills.id, skillId));
|
||||
const benchmarkDb = db;
|
||||
const service = companySkillService(db);
|
||||
for (const label of ["cold", "warm-1", "warm-2", "warm-new-service"]) {
|
||||
exporter.reset();
|
||||
rootPoolQueries = 0;
|
||||
const selected: ReturnType<typeof companySkillService> = label === "warm-new-service" ? companySkillService(db) : service;
|
||||
const runId = randomUUID();
|
||||
if (authenticated) await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", responsibleUserId: userId, contextSnapshot: { executionPolicy: {} } });
|
||||
rootPoolQueries = 0;
|
||||
const measuredStarted = performance.now();
|
||||
const resolve = () => selected.withRuntimeSkillEntries(companyId, {
|
||||
onlyKeys: new Set([key]), versionSelections: new Map([[key, version!.id]]), snapshotLatest: true,
|
||||
}, async entries => {
|
||||
if (authenticated) await benchmarkDb.update(heartbeatRuns).set({ contextSnapshot: sql`coalesce(${heartbeatRuns.contextSnapshot}, '{}'::jsonb) || ${JSON.stringify({ skillVersions: { [key]: version!.id }, skillSelectionSnapshot: true })}::jsonb` }).where(eq(heartbeatRuns.id, runId));
|
||||
return entries;
|
||||
});
|
||||
const entries: Awaited<ReturnType<typeof selected.listRuntimeSkillEntries>> = await withRunExecutionTiming(runId, () => createRunPreparationTiming(runId)("configuration_skills", () => authenticated
|
||||
? withSkillActor({ type: "agent", agentId, companyId, runId, source: "agent_jwt", onBehalfOfUserId: userId }, resolve)
|
||||
: resolve()));
|
||||
const elapsedMs = performance.now() - measuredStarted;
|
||||
const queryCount = rootPoolQueries;
|
||||
assert.equal(entries.length, 1);
|
||||
assert.equal(entries[0]!.sourceStatus, "available", entries[0]!.missingDetail ?? "Skill unavailable");
|
||||
// Verify the whole delivered repository, not just its authored SKILL.md.
|
||||
for (const file of projected) {
|
||||
const location = path.join(entries[0]!.source, file.path);
|
||||
assert.deepEqual(await fs.readFile(location), skillFileBytes(file));
|
||||
if (process.platform !== "win32") assert.equal(Boolean((await fs.stat(location)).mode & 0o111), Boolean(file.executable));
|
||||
}
|
||||
await provider.forceFlush();
|
||||
if (exportOtlp) assert.ok(exportResults.length > 0 && exportResults.every(code => code === 0), "OTLP export failed");
|
||||
const spans = exporter.getFinishedSpans();
|
||||
assert.ok(spans.some(span => span.name === "skills.version.verify" && span.attributes["paperclip.execution.fileCount"] === projected.length));
|
||||
assert.equal(new Set(spans.map(span => span.spanContext().traceId)).size, 1);
|
||||
assert.ok(spans.every(span => Object.keys(span.attributes).every(key => ["paperclip.task.run.run_id", "paperclip.execution.fileCount", "paperclip.execution.bytes"].includes(key))));
|
||||
samples.push({ label, elapsedMs, rootPoolQueries: queryCount, traceId: spans[0]!.spanContext().traceId,
|
||||
phases: spans.filter(span => span.name.startsWith("skills.")).map(span => ({ name: span.name,
|
||||
durationMs: span.duration[0] * 1000 + span.duration[1] / 1e6,
|
||||
spanId: span.spanContext().spanId, parentSpanId: span.parentSpanContext?.spanId,
|
||||
attributes: span.attributes })) });
|
||||
console.error(JSON.stringify({ label, elapsedMs, rootPoolQueries: queryCount }));
|
||||
}
|
||||
console.log(JSON.stringify({ schema: "paperclip.native-skill-preparation-benchmark.v1",
|
||||
sourceRevision, sourceChangesHash, exportedOtlp: exportOtlp,
|
||||
source: snapshotFile ? "supplied immutable snapshot" : "synthetic repository",
|
||||
networkDelayMsPerDirection: networkDelayMs, authenticated, stagingShape,
|
||||
repositoryHash: skillSnapshotHash(files), repositoryFiles: files.length, runtimeFiles: projected.length,
|
||||
runtimeBytes: projected.reduce((sum, file) => sum + skillFileBytes(file).length, 0), samples }, null, 2));
|
||||
} finally {
|
||||
await db?.$client.end({ timeout: 1 });
|
||||
if (proxy) {
|
||||
for (const socket of connections) socket.destroy();
|
||||
const server = proxy;
|
||||
await new Promise<void>(resolve => server.close(() => resolve()));
|
||||
}
|
||||
await database?.cleanup();
|
||||
await provider?.shutdown();
|
||||
trace.disable(); context.disable(); manager?.disable();
|
||||
for (const [name, value] of Object.entries({ PAPERCLIP_HOME: previous.home, PAPERCLIP_INSTANCE_ID: previous.instance, OTEL_EXPORTER_OTLP_ENDPOINT: previous.endpoint })) {
|
||||
if (value === undefined) delete process.env[name]; else process.env[name] = value;
|
||||
}
|
||||
await fs.rm(home, { recursive: true, force: true });
|
||||
}
|
||||
@@ -0,0 +1,151 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { context, SpanStatusCode, trace } from "@opentelemetry/api";
|
||||
import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks";
|
||||
import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace";
|
||||
import { companies, companySkills, companySkillVersions, companySkillRepositorySnapshots, createDb,
|
||||
EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { createRunPreparationTiming, withRunExecutionTiming } from "../services/execution-preparation-timing.js";
|
||||
import { skillSnapshotHash } from "../services/skill-snapshot.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
// This test owns its in-memory tracer; do not start the application exporter.
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "");
|
||||
const { companySkillService } = await import("../services/company-skills.js");
|
||||
const { withSkillVersionCacheLock } = await import("../services/skill-version-cache.js");
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const describeDatabase = support.supported ? describe : describe.skip;
|
||||
|
||||
describeDatabase("runtime skill observability", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let home: string;
|
||||
const exporter = new InMemorySpanExporter();
|
||||
const lifecycle: string[] = [];
|
||||
const provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter }), {
|
||||
onStart: span => { lifecycle.push(`start:${span.name}`); },
|
||||
onEnd: span => { lifecycle.push(`end:${span.name}`); },
|
||||
forceFlush: async () => {}, shutdown: async () => {},
|
||||
}] });
|
||||
const manager = new AsyncLocalStorageContextManager();
|
||||
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("skill-observability-");
|
||||
db = createDb(database.connectionString);
|
||||
home = await fs.mkdtemp(path.join(os.tmpdir(), "skill-observability-"));
|
||||
vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test");
|
||||
expect(context.setGlobalContextManager(manager.enable())).toBe(true);
|
||||
expect(trace.setGlobalTracerProvider(provider)).toBe(true);
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
|
||||
afterAll(async () => {
|
||||
await provider.shutdown();
|
||||
trace.disable(); context.disable(); manager.disable();
|
||||
vi.unstubAllEnvs();
|
||||
await db?.$client.end({ timeout: 1 });
|
||||
await database?.cleanup();
|
||||
if (home) await fs.rm(home, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
function deferred() {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>(done => { resolve = done; });
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
async function fixture() {
|
||||
const companyId = randomUUID();
|
||||
const skillId = randomUUID();
|
||||
const key = `company/${companyId}/traced`;
|
||||
const files = [{ path: "skills/traced/SKILL.md", kind: "skill" as const,
|
||||
content: "---\nname: traced\ndescription: Trace fixture\n---\nReport the model ID.\n" }];
|
||||
await db.insert(companies).values({ id: companyId, name: "Trace fixture", issuePrefix: `TR${companyId.slice(0, 6)}` });
|
||||
await db.insert(companySkills).values({ id: skillId, companyId, key, slug: "traced", name: "Traced",
|
||||
sourceType: "github", sourceLocator: "https://example.invalid/immutable-fixture", markdown: files[0]!.content });
|
||||
const [snapshot] = await db.insert(companySkillRepositorySnapshots).values({ companyId,
|
||||
contentHash: skillSnapshotHash(files), fileInventory: files }).returning();
|
||||
const [version] = await db.insert(companySkillVersions).values({ companyId, companySkillId: skillId,
|
||||
revisionNumber: 1, repositorySnapshotId: snapshot!.id, repositorySkillPath: files[0]!.path }).returning();
|
||||
await db.update(companySkills).set({ currentVersionId: version!.id }).where(eq(companySkills.id, skillId));
|
||||
const service = companySkillService(db);
|
||||
const options = { onlyKeys: new Set([key]), versionSelections: new Map([[key, version!.id]]), snapshotLatest: true };
|
||||
return { companyId, service, options };
|
||||
}
|
||||
|
||||
it("parents real preparation work and ends acquisition before the consumer settles", async () => {
|
||||
const f = await fixture();
|
||||
exporter.reset();
|
||||
lifecycle.length = 0;
|
||||
const entered = deferred(), release = deferred();
|
||||
const runId = randomUUID();
|
||||
const preparation = withRunExecutionTiming(runId, () => createRunPreparationTiming(runId)("configuration_skills",
|
||||
() => f.service.withRuntimeSkillEntries(f.companyId, f.options, async entries => {
|
||||
entered.resolve(); await release.promise; return entries;
|
||||
})));
|
||||
try {
|
||||
await Promise.race([entered.promise, preparation]);
|
||||
expect(exporter.getFinishedSpans().filter(span => span.name === "skills.cache.acquire")).toHaveLength(1);
|
||||
expect(exporter.getFinishedSpans().some(span => span.name === "skills.consume")).toBe(false);
|
||||
} finally { release.resolve(); await preparation; }
|
||||
expect(await preparation).toMatchObject([{ sourceStatus: "available" }]);
|
||||
await provider.forceFlush();
|
||||
const spans = exporter.getFinishedSpans();
|
||||
const span = (name: string) => { const found = spans.find(candidate => candidate.name === name); assert.ok(found, name); return found; };
|
||||
for (const [child, parent] of [
|
||||
["heartbeat.prepare.configuration_skills", "task.execution_attempt"],
|
||||
["skills.inventory", "heartbeat.prepare.configuration_skills"],
|
||||
["skills.inventory.bundled", "skills.inventory"],
|
||||
["skills.inventory.releases", "skills.inventory"],
|
||||
["skills.inventory.folders", "skills.inventory"],
|
||||
["skills.inventory.local_sources", "skills.inventory"],
|
||||
["skills.cache.acquire", "heartbeat.prepare.configuration_skills"],
|
||||
["skills.rows", "heartbeat.prepare.configuration_skills"],
|
||||
["skills.version.read", "heartbeat.prepare.configuration_skills"],
|
||||
["skills.version.materialize", "heartbeat.prepare.configuration_skills"],
|
||||
["skills.repository.read", "skills.version.materialize"],
|
||||
["skills.version.verify", "skills.version.materialize"],
|
||||
["skills.consume", "heartbeat.prepare.configuration_skills"],
|
||||
]) expect(span(child!).parentSpanContext?.spanId, child).toBe(span(parent!).spanContext().spanId);
|
||||
const acquisition = span("skills.cache.acquire");
|
||||
expect(acquisition.status.code).not.toBe(SpanStatusCode.ERROR);
|
||||
// SDK timestamps can differ by less than a millisecond at this boundary.
|
||||
// Observe actual lifecycle order instead of comparing rounded wall clocks.
|
||||
expect(lifecycle.indexOf("end:skills.cache.acquire")).toBeLessThan(lifecycle.indexOf("start:skills.rows"));
|
||||
expect(new Set(spans.map(value => value.spanContext().traceId)).size).toBe(1);
|
||||
expect(spans.every(value => Object.keys(value.attributes).every(key => [
|
||||
"paperclip.task.run.run_id", "paperclip.execution.fileCount", "paperclip.execution.bytes",
|
||||
].includes(key)))).toBe(true);
|
||||
}, 30_000);
|
||||
|
||||
it("ends a failed real lock acquisition and preserves the database error", async () => {
|
||||
const f = await fixture();
|
||||
await f.service.listRuntimeSkillEntries(f.companyId, f.options);
|
||||
const name = new URL(database.connectionString).pathname.slice(1);
|
||||
assert.match(name, /^[a-zA-Z0-9_-]+$/);
|
||||
await db.$client.unsafe(`ALTER DATABASE "${name}" SET lock_timeout TO '100ms'`);
|
||||
const entered = deferred(), release = deferred();
|
||||
const holder = withSkillVersionCacheLock(db, f.companyId, async () => { entered.resolve(); await release.promise; });
|
||||
try {
|
||||
await Promise.race([entered.promise, holder]);
|
||||
exporter.reset();
|
||||
const runId = randomUUID();
|
||||
await expect(withRunExecutionTiming(runId, () => createRunPreparationTiming(runId)("configuration_skills",
|
||||
() => f.service.listRuntimeSkillEntries(f.companyId, f.options)))).rejects.toMatchObject({ cause: { code: "55P03" } });
|
||||
await provider.forceFlush();
|
||||
const spans = exporter.getFinishedSpans();
|
||||
const acquisitions = spans.filter(span => span.name === "skills.cache.acquire");
|
||||
expect(acquisitions).toHaveLength(1);
|
||||
expect(acquisitions[0]!.status.code).toBe(SpanStatusCode.ERROR);
|
||||
const configuration = spans.find(span => span.name === "heartbeat.prepare.configuration_skills");
|
||||
expect(acquisitions[0]!.parentSpanContext?.spanId).toBe(configuration?.spanContext().spanId);
|
||||
expect(spans.some(span => span.name === "skills.rows")).toBe(false);
|
||||
} finally { release.resolve(); await holder; }
|
||||
expect(await f.service.listRuntimeSkillEntries(f.companyId, f.options)).toMatchObject([{ sourceStatus: "available" }]);
|
||||
}, 30_000);
|
||||
});
|
||||
@@ -842,6 +842,8 @@ describe("connection management with production default agents", () => {
|
||||
});
|
||||
const path = `/chat-endpoints/${created.body.id}/slack/test-message`;
|
||||
if (rate) {
|
||||
// Keep both calls in one rate window even when CI crosses a minute boundary.
|
||||
vi.useFakeTimers({ toFake: ["Date"] });
|
||||
// Shared buckets admit one actual send, while each principal's own
|
||||
// limit also prevents the next send. Neither may consume twice.
|
||||
expect(await f.request("POST", path, { requestId: randomUUID(), slackUserId: "UHUMAN" }, f.jwt), mode)
|
||||
@@ -856,7 +858,7 @@ describe("connection management with production default agents", () => {
|
||||
expect(sends, mode).toBe(rate ? 1 : 0);
|
||||
const audit = await server.db.select().from(toolAccessAuditEvents).where(eq(toolAccessAuditEvents.companyId, f.companyId));
|
||||
expect(audit.some(event => event.details.agentId === f.agentId && event.actorId === f.agentId), mode).toBe(true);
|
||||
} finally { provider = null; }
|
||||
} finally { vi.useRealTimers(); provider = null; }
|
||||
}
|
||||
}, 90_000);
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { ensureSkillCacheDirectory, touchSkillVersionCache, withSkillVersionCacheLock } from "./skill-version-cache.js";
|
||||
import { measureRunPhase } from "./execution-preparation-timing.js";
|
||||
import { verifyExistingBundledSkillRelease } from "./bundled-skill-release-validation.js";
|
||||
import { retainProjectSkillAccess, skillAccessService } from "./skill-access.js";
|
||||
import type { AuthorizationActor } from "./authorization.js";
|
||||
@@ -3237,7 +3238,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
if (!visibility().canReadCompany(companyId)) return;
|
||||
const existingRefresh = skillInventoryRefreshPromises.get(companyId);
|
||||
if (existingRefresh) {
|
||||
await existingRefresh;
|
||||
await measureRunPhase("skills.inventory.wait", () => existingRefresh);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -3250,10 +3251,10 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
if (!companyExists) {
|
||||
throw notFound("Company not found");
|
||||
}
|
||||
const bundledSkills = await ensureBundledSkills(companyId);
|
||||
await ensureBundledSkillReleases(companyId, bundledSkills);
|
||||
await reconcilePaperclipSkillFolders(companyId);
|
||||
await reconcileLocalPathSkillSources(companyId);
|
||||
const bundledSkills = await measureRunPhase("skills.inventory.bundled", () => ensureBundledSkills(companyId));
|
||||
await measureRunPhase("skills.inventory.releases", () => ensureBundledSkillReleases(companyId, bundledSkills));
|
||||
await measureRunPhase("skills.inventory.folders", () => reconcilePaperclipSkillFolders(companyId));
|
||||
await measureRunPhase("skills.inventory.local_sources", () => reconcileLocalPathSkillSources(companyId));
|
||||
})();
|
||||
|
||||
skillInventoryRefreshPromises.set(companyId, refreshPromise);
|
||||
@@ -6009,8 +6010,9 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
|
||||
async function runtimeVersionInventory(companyId: string, version: CompanySkillVersion) {
|
||||
if (!version.repositorySnapshotId) return version.fileInventory;
|
||||
const [snapshot] = await db.select().from(companySkillRepositorySnapshots).where(and(
|
||||
eq(companySkillRepositorySnapshots.companyId, companyId), eq(companySkillRepositorySnapshots.id, version.repositorySnapshotId)));
|
||||
const snapshotId = version.repositorySnapshotId;
|
||||
const [snapshot] = await measureRunPhase("skills.repository.read", () => db.select().from(companySkillRepositorySnapshots).where(and(
|
||||
eq(companySkillRepositorySnapshots.companyId, companyId), eq(companySkillRepositorySnapshots.id, snapshotId))));
|
||||
if (!snapshot || !version.repositorySkillPath) throw unprocessable("Installed repository snapshot is unavailable.");
|
||||
return repositorySkillProjection(snapshot.fileInventory, version.repositorySkillPath);
|
||||
}
|
||||
@@ -6029,7 +6031,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
const skillDir = path.join(runtimeRoot, version.id);
|
||||
return runVersionSnapshotFlight(skillDir, async () => {
|
||||
await ensureSkillCacheDirectory(runtimeRoot, true);
|
||||
if (await materializedVersionSnapshotMatches(skillDir, version)) return skillDir;
|
||||
if (await measureRunPhase("skills.version.verify", () => materializedVersionSnapshotMatches(skillDir, version), { fileCount: version.fileInventory.length })) return skillDir;
|
||||
const staging = await fs.mkdtemp(path.join(runtimeRoot, `.${version.id}-`));
|
||||
try {
|
||||
let wroteSkillFile = false;
|
||||
@@ -6070,7 +6072,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
?? (options.snapshotLatest || (skill.metadata?.skillSourceId && skill.metadata?.snapshotHash) || skill.accessSourceId ? skill.currentVersionId : null);
|
||||
if (selectedVersionId) {
|
||||
const versionPath = path.resolve(resolveManagedSkillsRoot(companyId), "__versions__", skill.id, selectedVersionId);
|
||||
const version = await getVersion(companyId, skill.id, selectedVersionId);
|
||||
const version = await measureRunPhase("skills.version.read", () => getVersion(companyId, skill.id, selectedVersionId));
|
||||
if (!version) {
|
||||
return {
|
||||
status: "missing",
|
||||
@@ -6096,7 +6098,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
detail: "The selected skill version will be materialized on the next run." };
|
||||
}
|
||||
}
|
||||
const versionSource = await materializeVersionSnapshot(companyId, skill, version);
|
||||
const versionSource = await measureRunPhase("skills.version.materialize", () => materializeVersionSnapshot(companyId, skill, version));
|
||||
if (versionSource) return { status: "available", source: versionSource };
|
||||
return { status: "missing", source: versionPath, detail: "The selected skill version produced no files." };
|
||||
} catch (error) {
|
||||
@@ -6174,15 +6176,18 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
): Promise<T> {
|
||||
if (options.onlyKeys?.size === 0 || (!options.onlyKeys && options.selectionConfig && !options.includeLegacy
|
||||
&& readPaperclipSkillSyncPreference(options.selectionConfig).desiredSkills.length === 0)) return consume([]);
|
||||
await ensureSkillInventoryCurrent(companyId);
|
||||
return withSkillVersionCacheLock(db, companyId, async () => consume(await listRuntimeSkillEntriesLocked(companyId, options)));
|
||||
await measureRunPhase("skills.inventory", () => ensureSkillInventoryCurrent(companyId));
|
||||
return withSkillVersionCacheLock(db, companyId, async () => {
|
||||
const entries = await listRuntimeSkillEntriesLocked(companyId, options);
|
||||
return measureRunPhase("skills.consume", () => consume(entries));
|
||||
});
|
||||
}
|
||||
|
||||
async function listRuntimeSkillEntriesLocked(
|
||||
companyId: string,
|
||||
options: RuntimeSkillEntryOptions,
|
||||
): Promise<PaperclipSkillEntry[]> {
|
||||
const skills = await readFullSkillRows(companyId);
|
||||
const skills = await measureRunPhase("skills.rows", () => readFullSkillRows(companyId));
|
||||
const selectedKeys = options.onlyKeys ?? (options.selectionConfig ? new Set(
|
||||
(options.includeLegacy ? resolveLegacyPaperclipDesiredSkillNames : resolvePaperclipDesiredSkillNames)(
|
||||
options.selectionConfig, skills.map(skill => ({ key: skill.key, runtimeName: buildSkillRuntimeName(skill.key, skill.slug) }))),
|
||||
|
||||
@@ -41,7 +41,8 @@ it.each([undefined, "", " "])("does not consult tracing for endpoint %s", async
|
||||
const error = vi.spyOn(console, "error");
|
||||
try {
|
||||
await expect(createRunPreparationTiming("run", "native.controller")("backend_setup", () => 42)).resolves.toBe(42);
|
||||
for (const phase of ["execution.usage_startup", "execution.mcp_handoff", "execution.dispatch_admission", "execution.budget_reservation"] as const) {
|
||||
for (const phase of ["execution.usage_startup", "execution.mcp_handoff", "execution.dispatch_admission", "execution.budget_reservation",
|
||||
"skills.inventory", "skills.inventory.wait", "skills.cache.acquire", "skills.version.materialize", "skills.consume"] as const) {
|
||||
const result = {};
|
||||
const work = vi.fn(() => {
|
||||
recordRepositoryObjectCacheDetail({ outcome: "hit", referenceSupplied: true });
|
||||
|
||||
@@ -34,6 +34,10 @@ type PreparationPhase = "environment_selection" | "execution_configuration" | "s
|
||||
| "instruction_changes" | "warm_release" | "instruction_collection";
|
||||
|
||||
export type ExecutionPhase =
|
||||
| "skills.inventory" | "skills.inventory.wait" | "skills.inventory.bundled" | "skills.inventory.releases"
|
||||
| "skills.inventory.folders" | "skills.inventory.local_sources" | "skills.rows"
|
||||
| "skills.cache.acquire" | "skills.version.read" | "skills.repository.read"
|
||||
| "skills.version.verify" | "skills.version.materialize" | "skills.consume"
|
||||
| "workspace.transaction" | "workspace.transaction.entry" | "workspace.transaction.commit" | "workspace.transaction.rollback"
|
||||
| "workspace.identity_read" | "workspace.row_lock" | "workspace.provenance" | "workspace.notification" | "workspace.write"
|
||||
| "workspace.sources" | "workspace.sources.canonicalize" | "workspace.sources.inventory" | "workspace.sources.alias_scan"
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
createNativeRunTrace,
|
||||
nativeRunPreparationStarts,
|
||||
recordFailedSkillPreparation,
|
||||
recordNativeProviderPerformanceSpans,
|
||||
NATIVE_RUN_SPAN_EVENT_TYPE,
|
||||
NATIVE_RUN_TRACE_SCHEMA_VERSION,
|
||||
} from "./native-run-trace.js";
|
||||
@@ -66,6 +67,39 @@ function createRecordingTraceContext(): {
|
||||
}
|
||||
|
||||
describe("native runner performance trace", () => {
|
||||
it("projects a full durable diagnostic batch without waiting on duplicate run-log writes", async () => {
|
||||
const {traceContext, spans}=createRecordingTraceContext();
|
||||
let release!: () => void;
|
||||
const barrier = new Promise<void>(resolve => { release=resolve; });
|
||||
const write=vi.fn(async()=>barrier);
|
||||
const trace=createNativeRunTrace({runId:"provider-perf-blocked-log",startedAtMs:1000,traceContext,onEvent:write});
|
||||
const names=["provider.model.stream","provider.context.environment","provider.tools.resolve","provider.model.resolve",
|
||||
"provider.workspace.snapshot.track","provider.workspace.snapshot.patch","provider.workspace.snapshot.diff","provider.skills.discovery"];
|
||||
let completed=false;
|
||||
const work=recordNativeProviderPerformanceSpans(trace,Array.from({length:128},(_,index)=>({
|
||||
name:names[index%names.length],startedAtMs:1100+index,endedAtMs:1200+index,
|
||||
}))).then(()=>{completed=true;});
|
||||
try {
|
||||
await new Promise<void>(resolve=>setImmediate(resolve));
|
||||
expect(completed).toBe(true);
|
||||
expect(write).not.toHaveBeenCalled();
|
||||
expect(spans.filter(span=>span.parentName==="provider.runtime")).toHaveLength(128);
|
||||
} finally {release();await work;await trace.finish("ok");}
|
||||
});
|
||||
it("groups normalized provider phases under their real native execution parent", async () => {
|
||||
const {traceContext, spans}=createRecordingTraceContext();
|
||||
const trace=createNativeRunTrace({runId:"provider-perf",startedAtMs:1000,traceContext});
|
||||
const execute=trace.start("native.session.execute",{startedAtMs:1000});
|
||||
await recordNativeProviderPerformanceSpans(trace,[
|
||||
{name:"provider.context.environment",startedAtMs:1100,endedAtMs:1200},
|
||||
{name:"provider.model.stream",startedAtMs:1200,endedAtMs:1700},
|
||||
]);
|
||||
await trace.end(execute,{endedAtMs:1800});
|
||||
await trace.finish("ok");
|
||||
expect(spans.find(s=>s.name==="provider.runtime")).toMatchObject({parentName:"native.session.execute",endedAtMs:1700});
|
||||
expect(spans.filter(s=>s.name==="provider.model.stream"||s.name==="provider.context.environment").map(s=>s.parentName))
|
||||
.toEqual(["provider.runtime","provider.runtime"]);
|
||||
});
|
||||
it.each([7_200_000, 10_800_000])(
|
||||
"times the current answer after %i ms without charging prior questions or human wait",
|
||||
async (answeredAtMs) => {
|
||||
|
||||
@@ -138,6 +138,7 @@ interface NativeRunSpanScopeState extends NativeRunSpanScope {
|
||||
context: unknown;
|
||||
attributes: Record<string, string | number | boolean>;
|
||||
ended: boolean;
|
||||
persistRunLog: boolean;
|
||||
}
|
||||
|
||||
const NOOP_SPAN: SpanHandle = {
|
||||
@@ -323,6 +324,8 @@ export function createNativeRunTrace(input: {
|
||||
parentName?: string;
|
||||
startedAtMs?: number;
|
||||
attributes?: Record<string, string | number | boolean>;
|
||||
/** False only when an existing durable diagnostic batch already contains the measurements. */
|
||||
persistRunLog?: boolean;
|
||||
} = {},
|
||||
): NativeRunSpanScope => {
|
||||
const startedAtMs = finiteMs(options.startedAtMs ?? Date.now());
|
||||
@@ -364,6 +367,7 @@ export function createNativeRunTrace(input: {
|
||||
context,
|
||||
attributes,
|
||||
ended: false,
|
||||
persistRunLog: options.persistRunLog !== false,
|
||||
};
|
||||
scopes.add(state);
|
||||
openScopes.add(state);
|
||||
@@ -413,7 +417,7 @@ export function createNativeRunTrace(input: {
|
||||
});
|
||||
if (outcome === "failed") { try { state.span.setStatus({ code: 2 }); } catch {} }
|
||||
try { state.span.end(endedAtMs); } catch {}
|
||||
await emitCompletedSpanEvent({
|
||||
if (state.persistRunLog) await emitCompletedSpanEvent({
|
||||
name: state.name,
|
||||
parentName: state.parentName,
|
||||
startedAtMs: state.startedAtMs,
|
||||
@@ -458,11 +462,12 @@ export function createNativeRunTrace(input: {
|
||||
);
|
||||
};
|
||||
|
||||
const record = async (span: NativeRunHistoricalSpan): Promise<void> => {
|
||||
const record = async (span: NativeRunHistoricalSpan, options: { persistRunLog?: boolean } = {}): Promise<void> => {
|
||||
const scope = start(span.name, {
|
||||
parentName: span.parentName,
|
||||
startedAtMs: span.startedAtMs,
|
||||
attributes: span.attributes,
|
||||
persistRunLog: options.persistRunLog,
|
||||
});
|
||||
await end(scope, {
|
||||
endedAtMs: span.endedAtMs,
|
||||
@@ -528,6 +533,22 @@ export function createNativeRunTrace(input: {
|
||||
|
||||
export type NativeRunTrace = ReturnType<typeof createNativeRunTrace>;
|
||||
|
||||
/** Group already-normalized provider measurements without exporting provider attributes. */
|
||||
export async function recordNativeProviderPerformanceSpans(trace: NativeRunTrace, spans: NativeRunHistoricalSpan[]): Promise<void> {
|
||||
if (!spans.length) return;
|
||||
try {
|
||||
const scope = trace.start("provider.runtime", { parentName: "native.session.execute",
|
||||
startedAtMs: Math.min(...spans.map(span => span.startedAtMs)), persistRunLog: false });
|
||||
try {
|
||||
for (const span of spans) await trace.record({ name: span.name, parentName: "provider.runtime",
|
||||
startedAtMs: span.startedAtMs, endedAtMs: span.endedAtMs, outcome: span.outcome }, { persistRunLog: false });
|
||||
} finally {
|
||||
await trace.end(scope, { endedAtMs: Math.max(...spans.map(span => span.endedAtMs)),
|
||||
outcome: spans.some(span => span.outcome === "failed") ? "failed" : "ok" });
|
||||
}
|
||||
} catch { /* Diagnostics cannot replace the work's result. */ }
|
||||
}
|
||||
|
||||
/** Emit preparation failure even when execution aborts before a native session exists. */
|
||||
export async function recordFailedSkillPreparation(input: {
|
||||
runId: string;
|
||||
|
||||
@@ -17,6 +17,7 @@ import { assertExternalNativeRunAuthority } from "./external-native-authority.js
|
||||
import { stageRemoteLaunchAssets } from "./remote-launch-assets.js";
|
||||
import { createRunPreparationTiming } from "../execution-preparation-timing.js";
|
||||
import { configuredEnvironment } from "../../vendor/paperclip-runner/index.js";
|
||||
import { parseProviderPerformanceSpans } from "../../vendor/paperclip-runner/index.js";
|
||||
import { apiKeyBillingType } from "@paperclipai/adapter-utils/api-billing-type";
|
||||
import { agents } from "@paperclipai/db";
|
||||
import { isCloudManagedInstance } from "../cloud-instance.js";
|
||||
@@ -204,6 +205,7 @@ import {
|
||||
createNativeRunTrace,
|
||||
isNativeRunRootHistoricalSpan,
|
||||
nativeRunPreparationStarts,
|
||||
recordNativeProviderPerformanceSpans,
|
||||
type NativeRunHistoricalSpan,
|
||||
type NativeRunSpanScope,
|
||||
type NativeRunTrace,
|
||||
@@ -8527,6 +8529,9 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
assertControllerActive,
|
||||
assertRunAdmissionActive,
|
||||
onCommittedEvent: async (event) => {
|
||||
if (event.eventType === "harness.diagnostic") {
|
||||
await recordNativeProviderPerformanceSpans(trace, parseProviderPerformanceSpans(event.payload, Date.now()));
|
||||
}
|
||||
const textKind = !firstTokenRecorded ? observeDeliverableText(event) : null;
|
||||
if (textKind) {
|
||||
firstTokenRecorded = true;
|
||||
|
||||
@@ -10,6 +10,7 @@ import { assertSkillSnapshotPath } from "./skill-snapshot.js";
|
||||
import { repositorySkillFrontmatter } from "./skill-repository-package.js";
|
||||
import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared";
|
||||
import { parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js";
|
||||
import { beginRunPhase } from "./execution-preparation-timing.js";
|
||||
|
||||
export const SKILL_VERSION_CACHE_IDLE_MS = 60 * 60 * 1000;
|
||||
const locks = new AsyncLocalStorage<Array<{ db: Db; companyId: string }>>();
|
||||
@@ -26,14 +27,19 @@ export async function withSkillVersionCacheLock<T>(db: Db, companyId: string, ac
|
||||
if (held.some(lock => lock.db === db && lock.companyId === companyId)) return action();
|
||||
// Preparation uses the ordinary pool for version/pin writes. Borrowing a pool
|
||||
// connection for this lock would deadlock installations with a one-slot pool.
|
||||
const acquired = beginRunPhase("skills.cache.acquire");
|
||||
return withDedicatedDbConnection(db, connection => connection.transaction(async tx => {
|
||||
const key = `skill_version_cache:${companyId}`;
|
||||
if (tryOnly) {
|
||||
const rows = await tx.execute<{ acquired: boolean }>(sql`select pg_try_advisory_xact_lock(hashtextextended(${key}, 0)) as acquired`);
|
||||
if (!rows[0]?.acquired) throw new SkillCacheBusy();
|
||||
} else await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${key}, 0))`);
|
||||
acquired();
|
||||
return locks.run([...held, { db, companyId }], action);
|
||||
}));
|
||||
})).catch(error => {
|
||||
acquired(true);
|
||||
throw error;
|
||||
});
|
||||
}
|
||||
|
||||
export function skillVersionCachePaths(companyId: string, skillId: string, versionId: string) {
|
||||
|
||||
+1
@@ -154,6 +154,7 @@ export const resolveSourceCodexHome = runner.resolveSourceCodexHome;
|
||||
export const resolveCodexCommand = runner.resolveCodexCommand;
|
||||
export const resolvePinnedCodexCommand = runner.resolvePinnedCodexCommand;
|
||||
export const validatePrpEvent = runner.validatePrpEvent;
|
||||
export const parseProviderPerformanceSpans = runner.parseProviderPerformanceSpans;
|
||||
export const validatePrpStructuredRunResult =
|
||||
runner.validatePrpStructuredRunResult;
|
||||
|
||||
|
||||
Reference in new issue
Block a user