+
+ {collecting
+ ? "Collecting projects…"
+ : loaded
+ ? "No current changes"
+ : "Loading projects…"}
+
+ {loaded && !collecting && (
Agent hooks and diffx sync publish changed checkouts here.
diff --git a/apps/web/src/session-context.tsx b/apps/web/src/session-context.tsx
index 946f157a..43697186 100644
--- a/apps/web/src/session-context.tsx
+++ b/apps/web/src/session-context.tsx
@@ -191,24 +191,30 @@ export function SessionProvider({ children }: { children: ReactNode }) {
);
const { remove: removeHead, retry: reloadHeads } = heads;
// Syncs the local checkouts; null when it fails or one is running.
- const collectLatest = useCallback(async () => {
- if (collectingRef.current) return null;
- collectingRef.current = true;
- setCollecting(true);
- setCollectError("");
- try {
- const { data, error } = await api.POST("/api/v2/collections");
- if (!data) throw new Error(errorDetail(error, "Unable to sync changes"));
- reloadHeads();
- return data;
- } catch (error) {
- setCollectError(errorDetail(error, "Unable to sync changes"));
- return null;
- } finally {
- collectingRef.current = false;
- setCollecting(false);
- }
- }, [reloadHeads]);
+ const collectLatest = useCallback(
+ async (initial = false) => {
+ if (collectingRef.current) return null;
+ collectingRef.current = true;
+ setCollecting(true);
+ setCollectError("");
+ try {
+ const { data, error } = await api.POST("/api/v2/collections", {
+ params: { query: { initial } },
+ });
+ if (!data)
+ throw new Error(errorDetail(error, "Unable to sync changes"));
+ reloadHeads();
+ return data;
+ } catch (error) {
+ setCollectError(errorDetail(error, "Unable to sync changes"));
+ return null;
+ } finally {
+ collectingRef.current = false;
+ setCollecting(false);
+ }
+ },
+ [reloadHeads],
+ );
const headsRef = useRef(heads.contexts);
useEffect(() => {
headsRef.current = heads.contexts;
@@ -283,7 +289,11 @@ export function SessionProvider({ children }: { children: ReactNode }) {
if (!collectOnLoad || collected.current === route) return;
// Effect replay must not treat the in-flight collection as a failed one.
collected.current = route;
- void collectLatest().then((result) => {
+ void collectLatest(true).then((result) => {
+ if (result?.failures.length)
+ setCollectError(
+ `Unable to sync ${result.failures.length} ${result.failures.length === 1 ? "checkout" : "checkouts"}: ${result.failures.map((failure) => `${failure.path} (${failure.detail})`).join(", ")}`,
+ );
const contexts = result?.contexts ?? [];
const head =
route.kind === "live"
@@ -386,6 +396,8 @@ export function SessionProvider({ children }: { children: ReactNode }) {
// The root URL keeps waiting for changed work while showing the empty state.
const empty =
heads.loaded && !selectedId && route.kind === "root" && !collectOnLoad;
+ const statusError =
+ state.status === "error" ? state.detail : heads.error || collectError;
useEffect(() => {
if (!restorePickerFocus.current || (!empty && state.status === "loading"))
return;
@@ -423,36 +435,36 @@ export function SessionProvider({ children }: { children: ReactNode }) {
- {missing
- ? "No current snapshot"
- : empty
- ? heads.contexts.length
- ? "No changes to review"
- : "No review contexts yet"
- : state.status === "error" || heads.error
- ? "Cannot load diffx"
+ {statusError
+ ? collectError
+ ? "Cannot sync changes"
+ : "Cannot load diffx"
+ : missing
+ ? "No current snapshot"
+ : empty
+ ? heads.contexts.length
+ ? "No changes to review"
+ : "No review contexts yet"
: collectOnLoad
? "Collecting latest changes"
: "Loading diffx"}
- {missing
- ? collectError ||
- "This stream's snapshots expired or were deleted. Pick current work from Projects."
- : empty
- ? remote
- ? "Create a collector token in settings, then run diffx sync to publish your changes."
- : "Run diffx to review a checkout, or diffx sync to publish remotely."
- : state.status === "error"
- ? state.detail
- : heads.error ||
- (collectOnLoad
- ? "Reading checkouts for new changes…"
- : "Reading review context…")}
+ {statusError ||
+ (missing
+ ? "This stream's snapshots expired or were deleted. Pick current work from Projects."
+ : empty
+ ? remote
+ ? "Create a collector token in settings, then run diffx sync to publish your changes."
+ : "Run diffx to review a checkout, or diffx sync to publish remotely."
+ : collectOnLoad
+ ? "Reading checkouts for new changes…"
+ : "Reading review context…")}
{empty && remote && (
)}
- {(state.status === "error" || heads.error) && (
+ {statusError && (
{
setState({ status: "loading" });
+ if (collectError) {
+ collected.current = null;
+ setCollectError("");
+ setMissing(false);
+ setUncollected(null);
+ }
reloadHeads();
setAttempt((current) => current + 1);
}}
diff --git a/apps/web/test/browser/local-workspace.spec.ts b/apps/web/test/browser/local-workspace.spec.ts
index 13b35132..c75c54ef 100644
--- a/apps/web/test/browser/local-workspace.spec.ts
+++ b/apps/web/test/browser/local-workspace.spec.ts
@@ -1783,7 +1783,7 @@ for (const mode of ["local", "remote"]) {
}),
);
let collections = 0;
- await page.route("**/api/v2/collections", (route) => {
+ await page.route("**/api/v2/collections*", (route) => {
collections++;
return route.fulfill({ json: { contexts: heads, failures: [] } });
});
@@ -1896,6 +1896,142 @@ for (const mode of ["local", "remote"]) {
});
}
+test("local first launch waits for collection with an empty catalog", async ({
+ page,
+}) => {
+ const state = await localWorkspace(page);
+ await page.route("**/api/v2/bootstrap", (route) =>
+ route.fulfill({
+ json: {
+ mode: "local",
+ signupEnabled: false,
+ user: null,
+ csrfToken: null,
+ collect: true,
+ },
+ }),
+ );
+ const snapshot = {
+ ...state.context,
+ streamId: "local-stream",
+ latestContextId: state.context.id,
+ };
+ let heads: ApiContext[] = [];
+ await page.route("**/api/v2/contexts?*", (route) =>
+ route.fulfill({ json: { contexts: heads, nextCursor: null } }),
+ );
+ let collections = 0;
+ let pending: Route | undefined;
+ await page.route("**/api/v2/collections*", (route) => {
+ collections++;
+ expect(new URL(route.request().url()).searchParams.get("initial")).toBe(
+ "true",
+ );
+ pending = route;
+ });
+
+ await page.goto("/");
+ const status = page.locator("main.session-status");
+ await expect(status).toContainText("Collecting latest changes");
+ await expect(status).toHaveAttribute("aria-busy", "true");
+ const projects = page.getByRole("navigation", { name: "Projects" });
+ await expect(projects).toContainText("Collecting projects…");
+ await expect(projects).not.toContainText("No current changes");
+ await expect(projects).not.toContainText("Agent hooks and diffx sync");
+ await expect(
+ page.getByRole("heading", { name: "No review contexts yet" }),
+ ).toHaveCount(0);
+ await expect(page.locator("#viewer")).toHaveCount(0);
+ await expect.poll(() => collections).toBe(1);
+ if (!pending) throw new Error("Missing startup collection");
+ heads = [snapshot];
+ await pending.fulfill({ json: { contexts: heads, failures: [] } });
+ await ready(page);
+ await expect(page).toHaveURL(/\/streams\/local-stream$/);
+ expect(collections).toBe(1);
+});
+
+for (const path of ["/", "/streams/local-stream"]) {
+ test(`local startup collection failure retries from ${path} without stored reviews`, async ({
+ page,
+ }) => {
+ const state = await localWorkspace(page);
+ await page.route("**/api/v2/bootstrap", (route) =>
+ route.fulfill({
+ json: {
+ mode: "local",
+ signupEnabled: false,
+ user: null,
+ csrfToken: null,
+ collect: true,
+ },
+ }),
+ );
+ let heads: ApiContext[] = [];
+ await page.route("**/api/v2/contexts?*", (route) =>
+ route.fulfill({ json: { contexts: heads, nextCursor: null } }),
+ );
+ let collections = 0;
+ let pending: Route | undefined;
+ await page.route("**/api/v2/collections*", (route) => {
+ collections++;
+ if (collections === 1)
+ return route.fulfill({
+ status: 422,
+ json: { detail: "Unable to collect checkouts: git failed" },
+ });
+ pending = route;
+ });
+ await page.goto(path);
+ const status = page.locator("main.session-status");
+ await expect(status).toContainText("Cannot sync changes");
+ await expect(status).toContainText("git failed");
+ await expect(status).toHaveAttribute("aria-busy", "false");
+ await status.getByRole("button", { name: "Retry", exact: true }).click();
+ await expect(status).toContainText("Collecting latest changes");
+ await expect(status).toHaveAttribute("aria-busy", "true");
+ await expect.poll(() => collections).toBe(2);
+ if (!pending) throw new Error("Missing retried collection");
+ heads = [{ ...state.context, streamId: "local-stream" }];
+ await pending.fulfill({ json: { contexts: heads, failures: [] } });
+ await ready(page);
+ await expect(page).toHaveURL(/\/streams\/local-stream$/);
+ await expect(page.getByRole("alert")).toHaveCount(0);
+ expect(collections).toBe(2);
+ });
+}
+
+test("local startup reports failed checkouts while opening collected changes", async ({
+ page,
+}) => {
+ const state = await localWorkspace(page);
+ await page.route("**/api/v2/bootstrap", (route) =>
+ route.fulfill({
+ json: {
+ mode: "local",
+ signupEnabled: false,
+ user: null,
+ csrfToken: null,
+ collect: true,
+ },
+ }),
+ );
+ await page.route("**/api/v2/collections*", (route) =>
+ route.fulfill({
+ json: {
+ contexts: [{ ...state.context, streamId: "local-stream" }],
+ failures: [{ path: "/workspace/broken", detail: "git failed" }],
+ },
+ }),
+ );
+ await page.goto("/");
+ await ready(page);
+ await expect(page).toHaveURL(/\/streams\/local-stream$/);
+ await expect(page.getByRole("alert")).toContainText(
+ "Unable to sync 1 checkout: /workspace/broken (git failed)",
+ );
+});
+
test("local live URLs sync the checkouts on every load", async ({ page }) => {
const state = await localWorkspace(page);
await page.route("**/api/v2/bootstrap", (route) =>
@@ -1930,7 +2066,7 @@ test("local live URLs sync the checkouts on every load", async ({ page }) => {
release = resolve;
});
let failNext = false;
- await page.route("**/api/v2/collections", async (route) => {
+ await page.route("**/api/v2/collections*", async (route) => {
collected++;
if (collected === 1) await held;
if (failNext)
@@ -2020,8 +2156,11 @@ test("local sync collects every checkout without replacing the open review", asy
});
const collections: string[] = [];
let fail = false;
- await page.route("**/api/v2/collections", (route) => {
+ await page.route("**/api/v2/collections*", (route) => {
collections.push(route.request().method());
+ expect(new URL(route.request().url()).searchParams.get("initial")).toBe(
+ "false",
+ );
if (fail)
return route.fulfill({
status: 422,
diff --git a/cmd/diffx/input.go b/cmd/diffx/input.go
index 2a9e640c..58d6a8dd 100644
--- a/cmd/diffx/input.go
+++ b/cmd/diffx/input.go
@@ -20,16 +20,15 @@ import (
)
type loadedInput struct {
- directory string
- mode string
- snapshot review.RepositoryDiff
- processed time.Duration
- captureID string
- contextID string
- mcpURL string
- remote bool
- checkouts int
- failed int
+ directory string
+ mode string
+ snapshot review.RepositoryDiff
+ processed time.Duration
+ captureID string
+ contextID string
+ mcpURL string
+ remote bool
+ collecting bool
}
func acquireInput(values options, stdin *os.File) (initialInput, error) {
diff --git a/cmd/diffx/local.go b/cmd/diffx/local.go
index 947dede1..17dd0f7b 100644
--- a/cmd/diffx/local.go
+++ b/cmd/diffx/local.go
@@ -49,11 +49,7 @@ func runLocal(ctx context.Context, arguments []string, stdin *os.File, stdout, s
collect := func(ctx context.Context) (daemon.Collection, error) {
return collectWorkspaces(ctx, values, input.Paths, stderr)
}
- initial, err := collect(ctx)
- if err != nil {
- return err
- }
- selected = daemon.Input{Collection: initial, Collect: collect}
+ selected = daemon.Input{Collect: collect}
} else {
request, err := collectInitialInput(ctx, values, input)
if err != nil {
@@ -88,22 +84,19 @@ func runLocal(ctx context.Context, arguments []string, stdin *os.File, stdout, s
return strings.EqualFold(strings.TrimSpace(reply.answer), "y") || strings.EqualFold(strings.TrimSpace(reply.answer), "yes"), nil
}
}, func(status daemon.Status, published daemon.Published) {
+ if input.Kind == "worktree" {
+ address := status.BrowserURL + "/"
+ writeStartup(stdout, loadedInput{mode: "git", directory: label, collecting: true, processed: time.Since(started), mcpURL: status.BrowserURL + "/mcp"}, address)
+ openBrowser(values, address, stderr)
+ return
+ }
submitted, ok := published.Newest()
if !ok {
return
}
address := status.BrowserURL + "/contexts/" + url.PathEscape(submitted.Context.ID)
loaded := submissionInput(submitted, time.Since(started))
- if input.Kind == "capture" {
- loaded.mode, loaded.directory = "pipe", ""
- } else {
- loaded.directory = label
- loaded.checkouts, loaded.failed = len(published.Submissions), len(published.Failures)
- loaded.snapshot.Files = nil
- for _, value := range published.Submissions {
- loaded.snapshot.Files = append(loaded.snapshot.Files, value.Snapshot.Files...)
- }
- }
+ loaded.mode, loaded.directory = "pipe", ""
for _, failure := range published.Failures {
fmt.Fprintf(stderr, " diffx: %s: %s\n", failure.Path, failure.Detail)
}
diff --git a/cmd/diffx/output.go b/cmd/diffx/output.go
index 53678cb0..e99b655b 100644
--- a/cmd/diffx/output.go
+++ b/cmd/diffx/output.go
@@ -61,16 +61,13 @@ func writeStartup(writer io.Writer, input loadedInput, urls ...string) {
}
fmt.Fprintf(writer, " %s\n", color.paint(color.bold+color.cyan, "diffx "+displayVersion()))
if input.directory != "" {
- fmt.Fprintf(writer, " %s %s\n", color.paint(color.dim, "collected from: "), input.directory)
- }
- fmt.Fprintf(writer, " %s %s\n", color.paint(color.dim, "mode:"), input.mode)
- if input.checkouts > 0 {
- summary := plural(input.checkouts, "checkout", "checkouts")
- if input.failed > 0 {
- summary += " · " + color.paint(color.yellow, fmt.Sprintf("%d failed", input.failed))
+ label := "collected from: "
+ if input.collecting {
+ label = "collecting from: "
}
- fmt.Fprintf(writer, " %s %s\n", color.paint(color.dim, "checkouts:"), summary)
+ fmt.Fprintf(writer, " %s %s\n", color.paint(color.dim, label), input.directory)
}
+ fmt.Fprintf(writer, " %s %s\n", color.paint(color.dim, "mode:"), input.mode)
for _, url := range urls {
fmt.Fprintf(writer, " %s %s\n", color.paint(color.dim, "url:"), color.paint(color.green, url))
}
@@ -84,15 +81,19 @@ func writeStartup(writer io.Writer, input loadedInput, urls ...string) {
fmt.Fprintf(writer, " MCP: %s\n", input.mcpURL)
}
fmt.Fprintln(writer, " ")
- elapsed := max(input.processed.Round(time.Millisecond).Milliseconds(), 1)
- fmt.Fprintf(writer, " %s %s\n", color.paint(color.dim, "diff statistics"), color.paint(color.magenta, fmt.Sprintf("%dms", elapsed)))
- fmt.Fprintf(writer, " %s\n", plural(files, "file changed", "files changed"))
- fmt.Fprintf(writer, " %s\n", color.paint(color.green, plural(additions, "addition", "additions")))
- fmt.Fprintf(writer, " %s\n", color.paint(color.red, plural(deletions, "deletion", "deletions")))
- if binaries > 0 {
- fmt.Fprintf(writer, " %s\n", plural(binaries, "binary file", "binary files"))
+ if input.collecting {
+ fmt.Fprintln(writer, " Collecting checkout snapshots in the background…")
+ } else {
+ elapsed := max(input.processed.Round(time.Millisecond).Milliseconds(), 1)
+ fmt.Fprintf(writer, " %s %s\n", color.paint(color.dim, "diff statistics"), color.paint(color.magenta, fmt.Sprintf("%dms", elapsed)))
+ fmt.Fprintf(writer, " %s\n", plural(files, "file changed", "files changed"))
+ fmt.Fprintf(writer, " %s\n", color.paint(color.green, plural(additions, "addition", "additions")))
+ fmt.Fprintf(writer, " %s\n", color.paint(color.red, plural(deletions, "deletion", "deletions")))
+ if binaries > 0 {
+ fmt.Fprintf(writer, " %s\n", plural(binaries, "binary file", "binary files"))
+ }
+ fmt.Fprintln(writer, " ")
}
- fmt.Fprintln(writer, " ")
guidance := "ctrl-c to stop."
if input.remote {
guidance = "snapshot submitted to remote server."
diff --git a/docs/architecture.md b/docs/architecture.md
index e8514c99..3cb68d74 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -10,7 +10,7 @@ canonical record of package boundaries and architectural decisions.
| Concern | Local `diffx [PATH]` | Remote `diffx-server` |
| ---------- | ------------------------------------------------------------------- | ---------------------------------------- |
| Collection | Discovered checkouts at start, live page loads and sync; stdin once | Separate finite sync or hook producer |
-| Admission | Direct application call before serving | Authenticated durable queue and worker |
+| Admission | Direct application call; startup runs after serving | Authenticated durable queue and worker |
| Lifetime | Foreground; authenticated singleton replacement | Foreground; graceful worker shutdown |
| Storage | One owner of persistent SQLite | One owner of persistent SQLite |
| Queries | Discovered checkouts or capture via REST/MCP | Same application services via REST/MCP |
@@ -31,16 +31,21 @@ observation independently; one failure is reported without blocking the others.
Synced snapshots never replace the open review, matching remote arrivals. Piped
and fixture processes have no collection route. Local queries are scoped to the
discovered checkouts, or to the capture's submission, because the persistent
-home-directory database retains every earlier run. The entry point supplies that
-scope and widens it before ingesting a newly discovered checkout; `contextservice`
-applies it to listing, counting, resolution and deletion, and out-of-scope
+home-directory database retains every earlier run. The entry point supplies an
+initially empty selection and widens it before ingesting a newly discovered
+checkout; `contextservice` applies it to listing, counting, resolution and deletion, and out-of-scope
contexts are not found. A scoped selection without checkouts matches nothing.
+Local listeners and the browser start before checkout discovery and collection. A server-owned task performs the
+initial collection; the first live-page collection request joins its result,
+including when the task finished before the browser opened. Explicit Sync always
+collects fresh snapshots. Collection failures leave the server running for retry,
+and shutdown cancels and joins collection before releasing database ownership.
Activity metadata is collected by the shared producer pipeline, never by catalog
reads. Both compositions project activity from the winning stream submission and
use it for current-work navigation; collection time and acceptance sequence still
choose the stream head. Workspace discovery and web-triggered collection remain
-local-only. There is no background sync, watcher or periodic collection.
+local-only. There is no watcher or periodic collection.
Private local control supports authenticated status/shutdown only. Replacement
uses lifecycle locking, a validated descriptor and graceful shutdown; a PID
diff --git a/docs/development.md b/docs/development.md
index 9326a17c..6b015a02 100644
--- a/docs/development.md
+++ b/docs/development.md
@@ -267,7 +267,12 @@ local checkout sync when bootstrap reports `collect`; loading `/` or a live
`/streams/{streamId}` URL, the header's “Sync all checkouts” icon and Alt+R call
`POST /api/v2/collections`. Load-time sync opens the stream's synced head, or the
most recently active changed one for `/`, and shows “Collecting latest changes” until it
-resolves. An explicit sync never replaces the open review: it reloads the heads,
+resolves. The listener opens before background startup collection. Load-time
+requests send `initial=true` to join startup once without collecting twice;
+explicit Sync requests always collect fresh snapshots. While discovery is pending,
+the process scope matches no stored contexts. Startup failures keep the server
+available and show Retry when no stored review can open. An explicit sync never
+replaces the open review: it reloads the heads,
so the newer-head notice, drawer dots and global unread action announce new
snapshots, and feedback reports synced, updated and failed checkouts. The global
action opens the most recently active unopened changed head. The drawer and
@@ -276,7 +281,7 @@ heads and unchanged submissions do not create unread prompts. Activity ordering
uses catalog `lastChangedAt`, falling back to captured collection time, with
stable stream/context identity ties. Repositories use their most active visible
stream. Startup and root selection follow the same ordering. Collection triggers
-are unchanged; no background sync or Git watcher is scheduled.
+are unchanged; no periodic collection or Git watcher is scheduled.
The “Switch review” picker first selects a repository
or Piped, then an immutable observation. Repository observations retain branch,
worktree, source/run, comparison, and collection-time metadata. Submission/session
diff --git a/docs/system.md b/docs/system.md
index d7270278..797e8c07 100644
--- a/docs/system.md
+++ b/docs/system.md
@@ -35,13 +35,15 @@ register paths for later server collection or maintain its own catalog.
The application core has two compositions. See [architecture.md](architecture.md)
for contracts, durability rules and migration details.
-`diffx [PATH]` is one foreground process: discover and collect the checkouts
-under its paths, ingest directly, and serve UI/REST/MCP. Omitted paths mean the
+`diffx [PATH]` is one foreground process: serve UI/REST/MCP immediately, then
+discover and collect the checkouts under its paths and ingest directly in a
+background task. The UI shows loading feedback while waiting for the first
+snapshots; failures remain visible with Retry. Omitted paths mean the
current directory; `--path` repeats and `--scan-depth` (default 8) bounds the
search. Captured snapshots remain fixed after edits and branch switches until the
user requests collection by loading a live URL or pressing Sync, which discovers
the paths again and ingests a new snapshot of every checkout; nothing watches Git.
-Piped input is collected once.
+Piped input is validated and collected once before serving.
The home-directory database survives process termination, but each process
serves only the checkouts it discovered (every branch and earlier run of them)
or its own piped capture, so other checkouts' history reappears only when reopened.
@@ -112,9 +114,10 @@ opens the stream's latest head when loaded or navigated to, while
live URLs; the picker opens pinned ones; piped snapshots have no stream and stay
pinned. A live URL resolves once, so later arrivals raise the notice instead of
swapping the open review. With local collection enabled, loading `/` or a live
-URL first syncs the checkouts, then opens the stream's synced head or, for `/`,
-the most recently active changed one; a failed sync opens the stored head and
-reports the error. Otherwise `/` opens the most recently active changed head's
+URL first syncs the checkouts, joining startup on the first load through
+`POST /api/v2/collections?initial=true`, then opens the stream's synced head or,
+for `/`, the most recently active changed one; a failed sync opens the stored
+head and reports the error. Otherwise `/` opens the most recently active changed head's
live URL. Startup selection and the local default MCP context use the same
activity ordering.
The drawer's search filters those loaded heads by the picker's metadata terms;
diff --git a/internal/contextservice/scope_test.go b/internal/contextservice/scope_test.go
index c6fa01a2..39fbb9cb 100644
--- a/internal/contextservice/scope_test.go
+++ b/internal/contextservice/scope_test.go
@@ -83,6 +83,7 @@ func TestScopeLimitsCatalogAndLookupsToSelectedCheckoutOrCapture(t *testing.T) {
{"checkout", ScopeFor(selected), []string{first.ID, later.ID}, []string{other.ID, capture.ID, otherCapture.ID}},
{"checkouts", ScopeFor(selected, linkedInput), []string{first.ID, other.ID, later.ID}, []string{capture.ID, otherCapture.ID}},
{"no checkouts", Scope{SourceID: "source"}, nil, []string{first.ID, other.ID, later.ID, capture.ID, otherCapture.ID}},
+ {"awaiting discovery", ScopeFor(), nil, []string{first.ID, other.ID, later.ID, capture.ID, otherCapture.ID}},
{"capture", ScopeFor(repeated), []string{capture.ID}, []string{first.ID, other.ID, later.ID, otherCapture.ID}},
{"unscoped", Scope{}, []string{first.ID, other.ID, capture.ID, otherCapture.ID, later.ID}, nil},
} {
diff --git a/internal/contextservice/service.go b/internal/contextservice/service.go
index bb2006b5..18215281 100644
--- a/internal/contextservice/service.go
+++ b/internal/contextservice/service.go
@@ -64,6 +64,7 @@ type Service struct {
// Scope limits a composition to its selected checkouts or one capture. The
// zero value is unscoped. Contexts outside the scope behave as not found.
type Scope struct {
+ selected bool
SourceID string
Checkouts []ingestion.Checkout
SubmissionID string
@@ -71,9 +72,9 @@ type Scope struct {
// ScopeFor selects the requests' checkouts, so later collections and branches
// of those checkouts stay visible. Captures without checkout identity select
-// their own submission.
+// their own submission. An empty selection matches nothing until widened.
func ScopeFor(inputs ...ingestion.Request) Scope {
- var scope Scope
+ scope := Scope{selected: true}
for _, input := range inputs {
scope = scope.with(input)
}
@@ -95,7 +96,7 @@ func (scope Scope) with(input ingestion.Request) Scope {
}
func (scope Scope) IsZero() bool {
- return scope.SourceID == "" && len(scope.Checkouts) == 0 && scope.SubmissionID == ""
+ return !scope.selected && scope.SourceID == "" && len(scope.Checkouts) == 0 && scope.SubmissionID == ""
}
func (scope Scope) selection(contextID string) ingestion.Selection {
diff --git a/internal/daemon/foreground.go b/internal/daemon/foreground.go
index 273233a9..6be0016c 100644
--- a/internal/daemon/foreground.go
+++ b/internal/daemon/foreground.go
@@ -20,8 +20,8 @@ type (
Published = serverapp.Published
)
-// Input is the initial collection and, for checkouts, how to collect them
-// again on explicit user request.
+// Input supplies a fixed capture or checkout collection to run in the
+// background at startup and again on explicit user request.
type Input struct {
Collection Collection
Collect func(context.Context) (Collection, error)
diff --git a/internal/daemon/server.go b/internal/daemon/server.go
index 6f3c8375..ae37181d 100644
--- a/internal/daemon/server.go
+++ b/internal/daemon/server.go
@@ -65,8 +65,8 @@ func Run(ctx context.Context, settings Settings, runtimeDir string, initial *ing
return runServer(ctx, cancel, settings, "", instanceID, collection, nil, published, log.New(os.Stderr, "diffx: ", log.LstdFlags))
}
-// runServer scopes queries to the initially collected checkouts or capture;
-// collect, when present, lets the user collect the checkouts again.
+// runServer scopes reads to the capture or discovered checkouts. Checkout
+// discovery and collection start after serving, with an initially empty scope.
func runServer(ctx context.Context, cancel context.CancelFunc, settings Settings, runtimeDir, instanceID string, initial Collection, collect func(context.Context) (Collection, error), ready func(Status, Published), logger *log.Logger) error {
settings = normalizeSettings(settings)
if settings.RetentionDays < 1 || settings.RetentionDays > 106751 {
@@ -153,7 +153,22 @@ func runServer(ctx context.Context, cancel context.CancelFunc, settings Settings
value.Observations, err = service.Count(ctx)
return value, err
}
- configuration := serverapp.Options{Collect: collect}
+ configuration := serverapp.Options{}
+ tasks := []serverapp.Task{serverapp.Prune(store.PruneExpired)}
+ if collect != nil {
+ collections := serverapp.NewCollections(service, collect, defaultContextID)
+ configuration.Collections = collections
+ tasks = append(tasks, collections.Initial(func(published Published, err error) {
+ if err != nil {
+ logger.Printf("initial collection: %v", err)
+ return
+ }
+ logger.Printf("collected %d checkouts (%d failed)", len(published.Submissions), len(published.Failures))
+ for _, failure := range published.Failures {
+ logger.Printf("%s: %s", failure.Path, failure.Detail)
+ }
+ }))
+ }
mux := serverapp.Handler(ctx, service, store, assets, defaultContextID, configuration)
webServer := &http.Server{Handler: publicRequests(mux, settings), ReadHeaderTimeout: 5 * time.Second, IdleTimeout: 60 * time.Second, ErrorLog: logger}
endpoints := []serverapp.Endpoint{{Server: webServer, Listener: listener}}
@@ -182,7 +197,6 @@ func runServer(ctx context.Context, cancel context.CancelFunc, settings Settings
return err
}
}
- tasks := []serverapp.Task{serverapp.Prune(store.PruneExpired)}
return serverapp.Run(ctx, endpoints, tasks, func() {
if ready != nil {
ready(status, published)
diff --git a/internal/daemon/server_test.go b/internal/daemon/server_test.go
index b4d18d9a..58631b28 100644
--- a/internal/daemon/server_test.go
+++ b/internal/daemon/server_test.go
@@ -5,16 +5,130 @@ import (
"errors"
"fmt"
"io"
+ "log"
"net"
+ "net/http"
"net/url"
+ osuser "os/user"
+ "path/filepath"
+ "strings"
+ "sync"
"testing"
"time"
+ "github.com/flexdinesh/diffx/internal/collector"
"github.com/flexdinesh/diffx/internal/contextservice"
+ "github.com/flexdinesh/diffx/internal/reviewstore"
)
const serverPatch = "diff --git a/value.txt b/value.txt\n--- a/value.txt\n+++ b/value.txt\n@@ -1 +1 @@\n-old\n+new\n"
+func TestCheckoutServerServesBeforeCollectionAndWaitsForCollectorShutdown(t *testing.T) {
+ state := filepath.Join(t.TempDir(), "state.db")
+ store, err := reviewstore.Open(state)
+ if err != nil {
+ t.Fatal(err)
+ }
+ processUser, err := osuser.Current()
+ if err != nil {
+ t.Fatal(err)
+ }
+ user, err := store.User(processUser.Uid, processUser.Username)
+ if err != nil {
+ t.Fatal(err)
+ }
+ input, err := collector.CollectPatch(t.Context(), serverPatch, t.TempDir(), collector.Options{SourceID: "prior-process", SubmissionID: "prior-capture"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ service := contextservice.New(store, user)
+ if _, err := service.Ingest(t.Context(), input); err != nil {
+ t.Fatal(err)
+ }
+ if err := errors.Join(service.Close(), store.Close()); err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ ready := make(chan Status, 1)
+ collectionCanceled := make(chan struct{})
+ release := make(chan struct{})
+ var releaseOnce sync.Once
+ done := make(chan error, 1)
+ stopped := make(chan struct{})
+ go func() {
+ defer close(stopped)
+ done <- runServer(ctx, cancel, Settings{Host: "127.0.0.1", Port: 0, State: state}, "", "async-test", Collection{},
+ func(ctx context.Context) (Collection, error) {
+ <-ctx.Done()
+ close(collectionCanceled)
+ <-release
+ return Collection{}, ctx.Err()
+ }, func(status Status, published Published) {
+ if len(published.Submissions) != 0 {
+ t.Error("ready must not wait for checkout collection")
+ }
+ ready <- status
+ }, log.New(io.Discard, "", 0))
+ }()
+ t.Cleanup(func() {
+ cancel()
+ releaseOnce.Do(func() { close(release) })
+ select {
+ case <-stopped:
+ case <-time.After(5 * time.Second):
+ t.Error("server did not stop")
+ }
+ })
+ var status Status
+ select {
+ case status = <-ready:
+ case err := <-done:
+ t.Fatalf("startup: %v", err)
+ case <-time.After(5 * time.Second):
+ t.Fatal("server readiness blocked on checkout collection")
+ }
+ client := &http.Client{Timeout: 2 * time.Second}
+ for _, expected := range []struct{ path, body string }{
+ {"/", " 1 {
+ fresh.SubmissionID = "explicit-sync"
+ fresh.Metadata.CollectedAt++
+ }
+ return serverapp.Collection{Requests: []ingestion.Request{fresh}}, nil
+ }, "")
+ completed := runInitialCollection(t, collections)
+ select {
+ case <-started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("initial collection did not start")
+ }
+ if finishedBeforeRequest {
+ close(release)
+ if err := awaitCollection(t, completed); err != nil {
+ t.Fatal(err)
+ }
+ }
+ first := httptest.NewRecorder()
+ requestDone := make(chan struct{})
+ go func() {
+ collections.ServeHTTP(first, httptest.NewRequest(http.MethodPost, "/api/v2/collections?initial=true", nil))
+ close(requestDone)
+ }()
+ if !finishedBeforeRequest {
+ close(release)
+ if err := awaitCollection(t, completed); err != nil {
+ t.Fatal(err)
+ }
+ }
+ select {
+ case <-requestDone:
+ case <-time.After(5 * time.Second):
+ t.Fatal("first request did not join startup")
+ }
+ firstContexts := collectionResponse(t, first)
+ if got := calls.Load(); got != 1 {
+ t.Fatalf("first browser request recollected: %d calls", got)
+ }
+ second := httptest.NewRecorder()
+ collections.ServeHTTP(second, httptest.NewRequest(http.MethodPost, "/api/v2/collections", nil))
+ secondContexts := collectionResponse(t, second)
+ if got := calls.Load(); got != 2 {
+ t.Fatalf("explicit sync did not recollect: %d calls", got)
+ }
+ if firstContexts[0].ID != secondContexts[0].ID {
+ t.Fatal("unchanged explicit sync should reuse the observation")
+ }
+ })
+ }
+}
+
+func TestInitialCollectionFailureIsReportedThenRetryCollectsAgain(t *testing.T) {
+ input := collectGit(t)
+ service, _ := collectionService(t)
+ var calls atomic.Int32
+ collections := serverapp.NewCollections(service, func(context.Context) (serverapp.Collection, error) {
+ if calls.Add(1) == 1 {
+ return serverapp.Collection{}, errors.New("checkout unavailable")
+ }
+ return serverapp.Collection{Requests: []ingestion.Request{input}}, nil
+ }, "")
+ completed := runInitialCollection(t, collections)
+ if err := awaitCollection(t, completed); err == nil || !strings.Contains(err.Error(), "checkout unavailable") {
+ t.Fatalf("initial failure: %v", err)
+ }
+ first := httptest.NewRecorder()
+ collections.ServeHTTP(first, httptest.NewRequest(http.MethodPost, "/api/v2/collections?initial=true", nil))
+ if first.Code != http.StatusUnprocessableEntity || !strings.Contains(first.Body.String(), "checkout unavailable") {
+ t.Fatalf("first request lost initial failure: %d %s", first.Code, first.Body.String())
+ }
+ if got := calls.Load(); got != 1 {
+ t.Fatalf("first request retried startup unexpectedly: %d calls", got)
+ }
+ retry := httptest.NewRecorder()
+ collections.ServeHTTP(retry, httptest.NewRequest(http.MethodPost, "/api/v2/collections", nil))
+ collectionResponse(t, retry)
+ if got := calls.Load(); got != 2 {
+ t.Fatalf("retry collection calls: %d", got)
+ }
+}
+
+func TestExplicitSyncCollectsFreshBeforeAnyBrowserInitialRequest(t *testing.T) {
+ input := collectGit(t)
+ service, _ := collectionService(t)
+ var calls atomic.Int32
+ collections := serverapp.NewCollections(service, func(context.Context) (serverapp.Collection, error) {
+ fresh := input
+ if calls.Add(1) > 1 {
+ fresh.SubmissionID = "explicit-sync"
+ fresh.Metadata.CollectedAt++
+ }
+ return serverapp.Collection{Requests: []ingestion.Request{fresh}}, nil
+ }, "")
+ completed := runInitialCollection(t, collections)
+ if err := awaitCollection(t, completed); err != nil {
+ t.Fatal(err)
+ }
+ response := httptest.NewRecorder()
+ collections.ServeHTTP(response, httptest.NewRequest(http.MethodPost, "/api/v2/collections", nil))
+ collectionResponse(t, response)
+ if got := calls.Load(); got != 2 {
+ t.Fatalf("explicit sync reused stale startup collection: %d calls", got)
+ }
+}
+
+func TestBrowserCancellationDoesNotCancelInitialCollection(t *testing.T) {
+ input := collectGit(t)
+ service, _ := collectionService(t)
+ started, release := make(chan struct{}), make(chan struct{})
+ var calls atomic.Int32
+ collections := serverapp.NewCollections(service, func(ctx context.Context) (serverapp.Collection, error) {
+ if calls.Add(1) > 1 {
+ return serverapp.Collection{Requests: []ingestion.Request{input}}, nil
+ }
+ close(started)
+ select {
+ case <-release:
+ return serverapp.Collection{Requests: []ingestion.Request{input}}, nil
+ case <-ctx.Done():
+ return serverapp.Collection{}, ctx.Err()
+ }
+ }, "")
+ completed := runInitialCollection(t, collections)
+ select {
+ case <-started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("initial collection did not start")
+ }
+ requestCtx, cancelRequest := context.WithCancel(t.Context())
+ cancelRequest()
+ collections.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/api/v2/collections?initial=true", nil).WithContext(requestCtx))
+ close(release)
+ if err := awaitCollection(t, completed); err != nil {
+ t.Fatalf("browser cancellation interrupted startup: %v", err)
+ }
+ response := httptest.NewRecorder()
+ collections.ServeHTTP(response, httptest.NewRequest(http.MethodPost, "/api/v2/collections?initial=true", nil))
+ collectionResponse(t, response)
+ if got := calls.Load(); got != 1 {
+ t.Fatalf("canceled browser consumed startup result: %d calls", got)
+ }
+}
+
+func TestCollectionScopeStartsEmptyAndPublishesOnlySelectedCheckouts(t *testing.T) {
+ input := collectGit(t)
+ service, store := collectionService(t)
+ unscoped := contextservice.New(store, reviewstore.User{ID: service.UserID(), Name: "collection-test"})
+ defer unscoped.Close()
+ unrelated := input
+ unrelated.Metadata.CheckoutKey = "unrelated-checkout"
+ unrelated.SubmissionID = "unrelated-submission"
+ stored, err := unscoped.Ingest(t.Context(), unrelated)
+ if err != nil {
+ t.Fatal(err)
+ }
+ collections := serverapp.NewCollections(service, func(context.Context) (serverapp.Collection, error) {
+ return serverapp.Collection{Requests: []ingestion.Request{input}}, nil
+ }, "")
+ assets := fstest.MapFS{"index.html": {Data: []byte("review UI")}}
+ handler := serverapp.Handler(t.Context(), service, store, assets, "", serverapp.Options{Collections: collections})
+ list := func() contextservice.Page {
+ t.Helper()
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/v2/contexts", nil))
+ if response.Code != http.StatusOK {
+ t.Fatalf("catalog: %d %s", response.Code, response.Body.String())
+ }
+ var page contextservice.Page
+ if err := json.Unmarshal(response.Body.Bytes(), &page); err != nil {
+ t.Fatal(err)
+ }
+ return page
+ }
+ if page := list(); len(page.Contexts) != 0 {
+ t.Fatalf("startup exposed prior unrelated reviews: %#v", page.Contexts)
+ }
+ completed := runInitialCollection(t, collections)
+ if err := awaitCollection(t, completed); err != nil {
+ t.Fatal(err)
+ }
+ if page := list(); len(page.Contexts) != 1 || page.Contexts[0].ID == stored.Context.ID {
+ t.Fatalf("publication widened beyond selected checkout: %#v", page.Contexts)
+ }
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/v2/contexts/"+stored.Context.ID, nil))
+ if response.Code != http.StatusNotFound {
+ t.Fatalf("unrelated review resolved: %d %s", response.Code, response.Body.String())
+ }
+}
diff --git a/packages/api/openapi.yaml b/packages/api/openapi.yaml
index c1068210..22433e35 100644
--- a/packages/api/openapi.yaml
+++ b/packages/api/openapi.yaml
@@ -429,6 +429,13 @@ paths:
post:
operationId: collectCheckouts
description: Local checkout mode only. Discovers the process's selected paths again and collects every checkout and linked worktree found, ingesting each as a fresh observation. Absent when the process serves piped input, a fixture or remote storage.
+ parameters:
+ - name: initial
+ in: query
+ description: Join the background startup collection on the first live page load instead of collecting twice. Once consumed, later loads collect fresh snapshots. Explicit Sync requests omit this parameter to always collect fresh snapshots.
+ schema:
+ type: boolean
+ default: false
responses:
"200":
description: The contexts holding the new collections, in ingestion order, and checkouts that could not be collected. Unchanged content reuses its existing review.
diff --git a/packages/api/src/schema.d.ts b/packages/api/src/schema.d.ts
index eae36121..6da78d98 100644
--- a/packages/api/src/schema.d.ts
+++ b/packages/api/src/schema.d.ts
@@ -1378,7 +1378,10 @@ export interface operations {
};
collectCheckouts: {
parameters: {
- query?: never;
+ query?: {
+ /** @description Join the background startup collection on the first live page load instead of collecting twice. Once consumed, later loads collect fresh snapshots. Explicit Sync requests omit this parameter to always collect fresh snapshots. */
+ initial?: boolean;
+ };
header?: never;
path?: never;
cookie?: never;
diff --git a/test/conformance/foreground_test.go b/test/conformance/foreground_test.go
index 272eb214..9454650b 100644
--- a/test/conformance/foreground_test.go
+++ b/test/conformance/foreground_test.go
@@ -212,6 +212,16 @@ func TestForegroundSnapshotReplacementAndPersistentHistory(t *testing.T) {
for time.Now().Before(deadline) {
status, err := client.Status(t.Context())
if err == nil && status.State == "running" && status.InstanceID != previous {
+ // Readiness precedes collection. Join the initial live-page request
+ // before asserting snapshots or making a later explicit Sync.
+ response, err := http.Post(status.BrowserURL+"/api/v2/collections?initial=true", "application/json", nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ response.Body.Close()
+ if response.StatusCode != http.StatusOK {
+ t.Fatalf("initial collection: %d", response.StatusCode)
+ }
return status
}
time.Sleep(20 * time.Millisecond)