diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index f0a63d61..13432eb6 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -23,11 +23,13 @@ jobs: timeout-minutes: 10 name: Schema drift check # Verifies the embedded internal/schema/ingest.v1.json matches - # tracebloc/data-ingestors' master. A green PR that silently - # diverges from upstream is a real correctness hazard — a - # customer's YAML could pass `tracebloc ingest validate` locally - # but be rejected by jobs-manager (or vice versa). Forcing the - # sync as a PR step keeps drift visible. + # tracebloc/data-ingestors at the PINNED ref (scripts/.data-ingestors-ref), + # not a floating branch. A green PR that silently diverges from the schema + # jobs-manager enforces is a real correctness hazard — a customer's YAML + # could pass `tracebloc ingest validate` locally but be rejected in-cluster + # (or vice versa). Pinning stops an unrelated upstream commit from redding + # every open CLI PR; adopting upstream is a deliberate SHA bump + re-sync + # (backend#1009). runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -77,6 +79,14 @@ jobs: # us from having to retrofit it later. run: go test -race -cover ./... + - name: Coverage floor (internal/cli, internal/submit must not rot) + # `go test -cover` above prints numbers but asserts nothing. This + # enforces a per-package floor on the two load-bearing, historically + # thin-tested packages (the money path + submit orchestration) so a + # test deletion can't silently drop coverage. Floors ratchet UP only — + # see scripts/coverage-floor.sh (backend#1009). + run: ./scripts/coverage-floor.sh + lint: timeout-minutes: 10 name: Lint diff --git a/README.md b/README.md index 22579fab..8cbd409c 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ The customer-facing CLI for the tracebloc declarative ingestion path. Wraps the **v0.3.0 is released** — the latest stable [release](https://github.com/tracebloc/cli/releases/latest), cut from `develop`. It builds on v0.2.0's guided `data ingest` and `dataset rm` with a new `dataset list` command plus home-screen / output polish (clearer copy, guided-first framing). The binary implements `version`, `completion`, `data validate`, `cluster info`, and the full `data ingest` / `dataset list` / `dataset rm` flow — local schema validation, cluster discovery, data staging, submission, and Job watching, end to end. -`data ingest` covers **9 of 10 task categories**: `image_classification`, `object_detection`, `keypoint_detection`, `text_classification`, `masked_language_modeling`, `tabular_classification`, `tabular_regression`, `time_series_forecasting`, and `time_to_event_prediction`. `semantic_segmentation` is pending mask-sidecar support upstream ([data-ingestors#136](https://github.com/tracebloc/data-ingestors/issues/136)); `instance_segmentation` is not yet implemented. +`data ingest` covers **9 of 10 task categories**: `image_classification`, `object_detection`, `keypoint_detection`, `text_classification`, `masked_language_modeling`, `tabular_classification`, `tabular_regression`, `time_series_forecasting`, and `time_to_event_prediction`. `semantic_segmentation` is pending mask-sidecar support upstream ([data-ingestors#136](https://github.com/tracebloc/data-ingestors/issues/136)). The release pipeline ships [`v0.3.0`](https://github.com/tracebloc/cli/releases/latest) as **cosign-signed, multi-arch binaries** — Linux (`amd64`, `arm64`, `386`, `arm`), macOS (`amd64`, `arm64`), and Windows (`amd64`, `arm64`) — each with `SHA256SUMS` and the install scripts. Install via [Customer experience](#customer-experience) or [build from source](#building-from-source). (A Homebrew tap and the `install.tracebloc.io` vanity URL are later follow-ups; the GitHub release URL serves installs today.) @@ -117,7 +117,7 @@ All v0.1 phases are merged: Beyond the original phases, `data ingest` was widened from image-classification-only to 9 of 10 modalities, and the test suite gained unit-coverage wins plus a kind-based integration harness for the real-I/O seams. -**v0.2.0** added a friendlier guided `data ingest` and `dataset rm` on the home screen (#44, #47). **v0.3.0** added the `dataset list` command (#53) plus home-screen / output-spacing polish and feedback-copy refinements (#52, #56). **Next:** cloud-source ingestion (S3/GCS/HTTPS) for datasets above the 1 GiB local cap; `semantic_segmentation` ([data-ingestors#136](https://github.com/tracebloc/data-ingestors/issues/136)) and `instance_segmentation`. Smaller follow-ups are tracked as [open issues](https://github.com/tracebloc/cli/issues). +**v0.2.0** added a friendlier guided `data ingest` and `dataset rm` on the home screen (#44, #47). **v0.3.0** added the `dataset list` command (#53) plus home-screen / output-spacing polish and feedback-copy refinements (#52, #56). **Next:** cloud-source ingestion (S3/GCS/HTTPS) for datasets above the 1 GiB local cap; `semantic_segmentation` ([data-ingestors#136](https://github.com/tracebloc/data-ingestors/issues/136)). Smaller follow-ups are tracked as [open issues](https://github.com/tracebloc/cli/issues). Epic: [tracebloc/client#147](https://github.com/tracebloc/client/issues/147). diff --git a/internal/cli/client.go b/internal/cli/client.go index 4277bb0a..516bb954 100644 --- a/internal/cli/client.go +++ b/internal/cli/client.go @@ -4,6 +4,7 @@ import ( "context" "crypto/rand" "encoding/hex" + "encoding/json" "errors" "fmt" "net/http" @@ -64,7 +65,14 @@ func newClientCreateCmd() *cobra.Command { cmd := &cobra.Command{ Use: "create", Short: "Provision a tracebloc client for this machine (auto-named; no flags required)", - Args: cobra.NoArgs, + // HIDDEN: provisioning is the installer's job — provision.sh calls this with + // zero flags (cli#137). It stays fully callable (including `--help`, so the + // installer's capability probe still works), but is kept off the user-facing + // surface: a human running `client create` STANDALONE mints a client the + // installer never deploys — an orphaned "phantom" (backend#970). Mirrors the + // hidden `list`; leaves `tracebloc client` showing only the user-useful `status`. + Hidden: true, + Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, _ []string) error { return runClientCreate(cmd.Context(), printerFor(cmd), clientPrompter(), clientCreateOpts{name: name, location: location, kubeconfigPath: kubeconfigPath, contextOverride: contextOverride, credentialFile: credentialFile, yes: yes}) @@ -356,9 +364,10 @@ func runClientCreate(ctx context.Context, p *ui.Printer, pr prompter, opts clien case http.StatusForbidden: return askAnAdmin(ctx, p, client, "provision a client", "provisioning") case http.StatusConflict: - // Per RFC C.3 the only 409 on POST /edge-device/ is cluster_conflict - // (R6): this cluster_id is bound to another account. - return &exitError{code: 1, err: errors.New(crossAccountConflictMsg)} + // A 409 on POST /edge-device/ is a cross-account cluster_conflict + // (R6) or a same-account cluster_in_use; conflictMessage picks the + // right guidance (and names the owner when the backend supplies it). + return &exitError{code: 1, err: errors.New(conflictMessage(ae))} } } return &exitError{code: 1, err: cerr} @@ -437,10 +446,49 @@ func runClientCreate(ctx context.Context, p *ui.Printer, pr prompter, opts clien } // crossAccountConflictMsg is the guidance shown when this cluster — or the client -// already live on it — belongs to a different tracebloc account. Shared by the -// create 409 (R6) and the R7 not-owned / anchor-taken refusals so they read alike. -const crossAccountConflictMsg = "this cluster is already registered to a different tracebloc account — " + - "sign in to that account, or ask your admin (cluster_conflict)" +// already live on it — belongs to another tracebloc account, and we don't have the +// owner's contact (a client-side refusal, or a backend without the owner_email in +// its 409 body). Contact-the-owner first (never "delete the cluster" — it isn't +// ours to wipe). conflictMessage() enriches this with the owner's email when the +// backend supplies it. +const crossAccountConflictMsg = "this cluster is already registered to another tracebloc account — " + + "ask its owner to release it, or sign in as that account (cluster_conflict)" + +// conflictMessage turns a provisioning 409 body into user guidance. Fix #2 has the +// backend distinguish a genuine cross-account conflict (cluster_conflict, now +// carrying the owner's contact email so the user knows who to ask) from a +// same-account live sibling (cluster_in_use — two clients fighting over one +// cluster). Degrades to the generic cross-account text when the body isn't +// parseable (a backend predating fix #2) or carries no owner. +func conflictMessage(ae *api.APIError) string { + var body struct { + Error string `json:"error"` + OwnerEmail string `json:"owner_email"` + HolderName string `json:"holder_name"` + } + if ae != nil { + _ = json.Unmarshal([]byte(ae.Body), &body) + } + switch body.Error { + case "cluster_in_use": + who := body.HolderName + if who == "" { + who = "another of your clients" + } + return fmt.Sprintf( + "another tracebloc client (%s) in your account is already live on this cluster — "+ + "offboard it first with `tracebloc delete`, or provision on a separate machine (cluster_in_use)", + who) + default: // cluster_conflict, or an unrecognized / empty body + if body.OwnerEmail != "" { + return fmt.Sprintf( + "this cluster is already registered to another tracebloc account (%s) — "+ + "ask them to release it, or sign in as that account (cluster_conflict)", + body.OwnerEmail) + } + return crossAccountConflictMsg + } +} // adoptLiveInClusterClient implements the RFC-0001 §7.2 / R7 adopt-backfill. It // discovers a tracebloc client already live on the target cluster and, when the @@ -465,13 +513,28 @@ func adoptLiveInClusterClient( ) (*api.ProvisionedClient, bool, error) { live, err := readInClusterClient(ctx, cluster.KubeconfigOptions{Path: opts.kubeconfigPath, Context: opts.contextOverride}) if err != nil { - // Best-effort: couldn't inspect the cluster for a live client. Fall through - // to a plain create (the backend's cluster_id get-or-create still applies). - ilog.Logf("in-cluster client discovery failed (non-fatal): %v", err) + // We couldn't inspect the cluster for an existing client. Whether that's + // safe to ignore depends on reachability: clusterID != "" means we DID read + // the cluster's kube-system UID over the same kubeconfig, so the cluster is + // reachable and this is an RBAC/transient read failure — NOT proof it's + // empty. Minting here could create a duplicate over a live client that then + // never deploys and permanently strands the cluster anchor (the phantom-1060 + // class). Fail closed. Only a genuinely unreachable cluster (clusterID == "", + // where the UID read failed too) falls through to a plain, non-anchored + // create — that mint stamps no anchor, so it can't orphan one. + if clusterID != "" { + ilog.Logf("in-cluster client discovery failed on a reachable cluster (failing closed): %v", err) + return nil, false, &exitError{code: 1, err: fmt.Errorf( + "couldn't check whether a tracebloc client is already running on this cluster (%w) — "+ + "provisioning now could mint a duplicate that never deploys and locks the cluster to it. "+ + "Re-run (if this was transient); if it persists, ensure your kubeconfig/context can list "+ + "deployments and secrets across namespaces. Diagnose with `tracebloc cluster doctor`", err)} + } + ilog.Logf("in-cluster client discovery skipped — cluster unreachable (non-fatal): %v", err) return nil, false, nil } if live == nil { - return nil, false, nil // fresh cluster — nothing installed to adopt + return nil, false, nil // reachable, nothing installed to adopt — a genuine fresh cluster } ilog.Logf("live in-cluster client: id=%s namespace=%s", live.ClientID, live.Namespace) @@ -518,8 +581,11 @@ func adoptLiveInClusterClient( var ae *api.APIError switch { case errors.As(perr, &ae) && ae.StatusCode == http.StatusConflict: - // Anchor already taken (write-once / bound elsewhere — R6). - return nil, false, &exitError{code: 1, err: errors.New(crossAccountConflictMsg)} + // Anchor held by another client: cross-account (cluster_conflict, now + // naming the owner) or a same-account live sibling (cluster_in_use). + // Fix #2's same-account reclaim means this no longer fires for a stale + // same-account holder — that path now succeeds (200). + return nil, false, &exitError{code: 1, err: errors.New(conflictMessage(ae))} case errors.As(perr, &ae) && ae.StatusCode == http.StatusForbidden: return nil, false, askAnAdmin(ctx, p, apiClient, "provision a client", "provisioning") } diff --git a/internal/cli/client_test.go b/internal/cli/client_test.go index a721d845..cb0b8477 100644 --- a/internal/cli/client_test.go +++ b/internal/cli/client_test.go @@ -283,11 +283,72 @@ func TestClientCreate_R7_CrossAccountRefuse(t *testing.T) { err := runClientCreate(context.Background(), ui.New(&bytes.Buffer{}), nil, clientCreateOpts{name: "box", location: "DE", yes: true}) - if err == nil || !strings.Contains(err.Error(), "different tracebloc account") { + if err == nil || !strings.Contains(err.Error(), "registered to another tracebloc account") { t.Errorf("want cross-account refusal, got %v", err) } } +// TestClientCreate_R7_DiscoveryErrorReachableFailsClosed: the cluster is REACHABLE +// (its kube-system UID read cleanly, clusterID != "") but in-cluster client discovery +// ERRORS (RBAC/transient List failure). We can't tell whether a client is already +// running, so minting would risk a duplicate that never deploys and strands the +// cluster anchor (the phantom-1060 class). Must fail closed — no mint, no backfill. +func TestClientCreate_R7_DiscoveryErrorReachableFailsClosed(t *testing.T) { + withClientBackend(t, func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodGet && r.URL.Path == "/edge-device/": + _, _ = w.Write([]byte(`[]`)) // account list (fetched before adopt) is allowed + default: + t.Errorf("unexpected %s %s — must fail closed before any mint/backfill", r.Method, r.URL.Path) + } + }) + stubClusterID(t, "uid-9", nil) // cluster reachable + stubInClusterClient(t, nil, errors.New("forbidden: cannot list deployments")) + + err := runClientCreate(context.Background(), ui.New(&bytes.Buffer{}), nil, + clientCreateOpts{name: "box", location: "DE", yes: true}) + if err == nil || !strings.Contains(err.Error(), "couldn't check whether a tracebloc client is already running") { + t.Errorf("want fail-closed error, got %v", err) + } +} + +// TestClientCreate_DiscoveryErrorUnreachableMintsNonAnchored: when the cluster is +// genuinely UNREACHABLE (the UID read failed too → clusterID == ""), a discovery +// error is not proof a client is running, and a non-anchored mint stamps no anchor +// so it can't orphan one. Provisioning must still proceed (the deliberate no-cluster +// fallback), minting with an empty cluster_id. Guards against over-tightening the +// fail-closed gate into the legitimate headless path. +func TestClientCreate_DiscoveryErrorUnreachableMintsNonAnchored(t *testing.T) { + var body api.CreateClientRequest + postCalled := false + withClientBackend(t, func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodGet && r.URL.Path == "/edge-device/": + _, _ = w.Write([]byte(`[]`)) + case r.Method == http.MethodPost && r.URL.Path == "/edge-device/": + postCalled = true + _ = json.NewDecoder(r.Body).Decode(&body) + w.WriteHeader(http.StatusCreated) + _, _ = w.Write([]byte(`{"id":8,"first_name":"box","username":"u-8","namespace":"box","location":"DE"}`)) + default: + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + }) + stubClusterID(t, "", errors.New("no cluster reachable")) + stubInClusterClient(t, nil, errors.New("no cluster reachable")) + + if err := runClientCreate(context.Background(), ui.New(&bytes.Buffer{}), nil, + clientCreateOpts{name: "box", location: "DE", yes: true}); err != nil { + t.Fatalf("unreachable-cluster provisioning must still mint (non-anchored): %v", err) + } + if !postCalled { + t.Fatal("expected a non-anchored mint when the cluster is unreachable") + } + if body.ClusterID != "" { + t.Errorf("cluster_id = %q, want empty (non-anchored mint)", body.ClusterID) + } +} + func TestClientCreate_RequiresLogin(t *testing.T) { t.Setenv("TRACEBLOC_CONFIG_DIR", t.TempDir()) // no config → not signed in err := runClientCreate(context.Background(), ui.New(&bytes.Buffer{}), nil, clientCreateOpts{name: "x", location: "DE", yes: true}) @@ -658,11 +719,57 @@ func TestClientCreate_ClusterConflict(t *testing.T) { }) stubClusterID(t, "uid-1", nil) err := runClientCreate(context.Background(), ui.New(&bytes.Buffer{}), nil, clientCreateOpts{name: "c", location: "DE", yes: true}) - if err == nil || !strings.Contains(err.Error(), "different tracebloc account") { + if err == nil || !strings.Contains(err.Error(), "registered to another tracebloc account") { t.Errorf("want a cluster_conflict error, got %v", err) } } +// TestClientCreate_ClusterConflict_RevealsOwnerEmail: fix #2 has the backend put the +// owning account's contact email in the cross-account 409 body; the CLI surfaces it +// so the user knows who to ask to release the cluster. +func TestClientCreate_ClusterConflict_RevealsOwnerEmail(t *testing.T) { + withClientBackend(t, func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodGet: + _, _ = w.Write([]byte(`[]`)) + case r.Method == http.MethodPost: + w.WriteHeader(http.StatusConflict) + _, _ = w.Write([]byte(`{"error":"cluster_conflict","cluster_id":"uid-1","owner_email":"owner@other.test"}`)) + } + }) + stubClusterID(t, "uid-1", nil) + err := runClientCreate(context.Background(), ui.New(&bytes.Buffer{}), nil, clientCreateOpts{name: "c", location: "DE", yes: true}) + if err == nil || !strings.Contains(err.Error(), "owner@other.test") { + t.Errorf("want the owner email surfaced, got %v", err) + } + if !strings.Contains(err.Error(), "ask them to release it") { + t.Errorf("want contact-the-owner guidance, got %v", err) + } +} + +// TestClientCreate_ClusterInUse: a same-account live sibling already holds the +// anchor (fix #2's cluster_in_use). The message names the live client and points at +// offboarding it — NOT a cross-account "different account" message. +func TestClientCreate_ClusterInUse(t *testing.T) { + withClientBackend(t, func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodGet: + _, _ = w.Write([]byte(`[]`)) + case r.Method == http.MethodPost: + w.WriteHeader(http.StatusConflict) + _, _ = w.Write([]byte(`{"error":"cluster_in_use","cluster_id":"uid-1","holder_client_id":42,"holder_name":"other-box"}`)) + } + }) + stubClusterID(t, "uid-1", nil) + err := runClientCreate(context.Background(), ui.New(&bytes.Buffer{}), nil, clientCreateOpts{name: "c", location: "DE", yes: true}) + if err == nil || !strings.Contains(err.Error(), "other-box") || !strings.Contains(err.Error(), "cluster_in_use") { + t.Errorf("want a cluster_in_use message naming the live client, got %v", err) + } + if strings.Contains(err.Error(), "another tracebloc account") { + t.Errorf("cluster_in_use must NOT read as a cross-account conflict, got %v", err) + } +} + func TestClientCreate_NoClusterAnchorWarns(t *testing.T) { var body api.CreateClientRequest withClientBackend(t, func(w http.ResponseWriter, r *http.Request) { @@ -1338,3 +1445,29 @@ func TestClientStatus_WaitCtrlCIsSilent(t *testing.T) { t.Errorf("Ctrl-C should exit silently (nil-inner exitError), got: %v", err) } } + +func TestClientSubcommandVisibility(t *testing.T) { + // `create` and `list` are installer-internal — Hidden so a user isn't invited to + // run them (a standalone `tracebloc client create` mints a client the installer + // never deploys, i.e. an orphaned phantom, backend#970). `status` stays + // user-visible. Hidden != disabled: all remain runnable (the installer still + // invokes create/list). + hidden := map[string]bool{} + runnable := map[string]bool{} + for _, c := range newClientCmd().Commands() { + hidden[c.Name()] = c.Hidden + runnable[c.Name()] = c.RunE != nil + } + if !hidden["create"] { + t.Error("client create must be Hidden (installer-internal; standalone mints a phantom)") + } + if !hidden["list"] { + t.Error("client list must stay Hidden") + } + if hidden["status"] { + t.Error("client status must stay user-visible") + } + if !runnable["create"] { + t.Error("hidden create must still be runnable (the installer invokes it)") + } +} diff --git a/internal/cli/coverage_test.go b/internal/cli/coverage_test.go index 76cfde76..850dd448 100644 --- a/internal/cli/coverage_test.go +++ b/internal/cli/coverage_test.go @@ -99,11 +99,18 @@ func TestClassifyPushOutcome(t *testing.T) { wantStat string wantCode int }{ - {"clean", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 10, InsertedRecords: 10}}}, nil, "succeeded", 0}, + {"clean", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 10, InsertedRecords: 10, APISentRecords: 10}}}, nil, "succeeded", 0}, {"partial", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 10, InsertedRecords: 7, FailedRecords: 3}}}, nil, "completed_with_failures", 9}, {"failed", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeFailed}}, nil, "failed", 9}, + {"unknown", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeUnknown}}, nil, "unknown", 9}, {"detached", &submit.Result{Submit: resp}, nil, "detached", 0}, + {"nil result", nil, nil, "detached", 0}, {"watch error", &submit.Result{Submit: resp}, &submit.WatchError{Err: errors.New("stream broke")}, "watch_error", 9}, + // The submit-side error buckets (exit 5 vs 8) the original matrix missed. + {"auth 401", nil, &submit.SubmitError{StatusCode: 401}, "auth_error", 5}, + {"auth 403", nil, &submit.SubmitError{StatusCode: 403}, "auth_error", 5}, + {"submit 500", nil, &submit.SubmitError{StatusCode: 500}, "submit_error", 8}, + {"submit 422", nil, &submit.SubmitError{StatusCode: 422}, "submit_error", 8}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { diff --git a/internal/cli/data.go b/internal/cli/data.go index b3921238..6794b4b8 100644 --- a/internal/cli/data.go +++ b/internal/cli/data.go @@ -449,8 +449,8 @@ other collaborators train against it without ever seeing the raw files.`)) // supported case push.IsKnown(a.Spec.Category): // A recognized category data ingest doesn't implement yet — image - // (semantic_segmentation / instance_segmentation) or text - // (causal_language_modeling). Routed here (not the default branch) so the + // (semantic_segmentation) or text (causal_language_modeling, seq2seq, + // …). Routed here (not the default branch) so the // user gets the registry's per-category pending-support reason, not a // misleading "unrecognized category". Supported categories were already // caught above, so IsKnown here means known-but-unsupported. @@ -758,6 +758,37 @@ other collaborators train against it without ever seeing the raw files.`)) return &exitError{code: 7, err: stageErr} } + // 10–12. The ingestion-run tail: mint token → port-forward → submit → + // classify → emit JSON → reclaim staging. Extracted so its + // outcome matrix (exit 5/8/9, JSON emission, and the + // must-NOT-reclaim-on-partial gate) is table-testable via the + // injected seams without a cluster (#1009). jsonEmitted flows + // back so the --output-json error defer above stays correct. + je, runErr := runIngestionRun(ctx, out, a, target, specBytes, spec) + jsonEmitted = je + return runErr +} + +// runIngestionRun is the money path's outcome tail. It mints the ingestor +// token, port-forwards to jobs-manager, POSTs the run, classifies the result +// into a status + process exit code (kept in lockstep by classifyPushOutcome), +// emits the machine-readable JSON in --output-json mode, and reclaims the +// staged source copy on a clean success only. +// +// Split out of runDataIngest purely for testability: the four cluster-touching +// steps go through package-level seams (mintIngestorTokenFn / +// portForwardJobsManagerFn / submitRunFn / cleanStagingFn), so a table test can +// drive the full classify → exit-code → JSON → reclaim matrix — including the +// "must NOT reclaim on partial failure" gate — without standing up a cluster +// (#1009). +// +// Returns jsonEmitted so runDataIngest's --output-json error defer knows +// whether a result object already reached stdout: the mint / port-forward +// failures return before the emit and rely on that defer; the submit path +// always emits. +func runIngestionRun(ctx context.Context, out io.Writer, a runDataIngestArgs, target *clusterTarget, specBytes []byte, spec map[string]any) (jsonEmitted bool, err error) { + resolved, cs, release, pvc := target.Resolved, target.Clientset, target.Release, target.PVC + // 10. Mint the SA token Phase 4 uses to authenticate the POST // to jobs-manager. Expiry is 1 hour (vs cluster info's 10 // min) because the full Phase 4 lifecycle — submit + watch @@ -770,10 +801,10 @@ other collaborators train against it without ever seeing the raw files.`)) a.Printer.Hintf("Submitting the run, then following along as tracebloc validates your data and loads it into the table — progress streams below.") a.Printer.Hintf("This follows the run for up to an hour; a longer run keeps going on its own (or start it with --detach and check back later).") } - tok, err := cluster.MintIngestorToken(ctx, cs, resolved.Namespace, + tok, err := mintIngestorTokenFn(ctx, cs, resolved.Namespace, release.IngestorSAName, 3600, nil) if err != nil { - return &exitError{code: 5, err: err} + return false, &exitError{code: 5, err: err} } // 11. Open a port-forward to a Pod backing the jobs-manager @@ -784,10 +815,10 @@ other collaborators train against it without ever seeing the raw files.`)) // `kubectl port-forward`. Bugbot PR #10 r3 caught the // original broken-by-design direct-URL POST. a.Printer.Infof("Connecting to your workspace to submit the run…") - pf, err := submit.PortForwardJobsManager(ctx, cs, resolved.RestConfig, + pf, err := portForwardJobsManagerFn(ctx, cs, resolved.RestConfig, resolved.Namespace, release.JobsManagerServiceName, release.JobsManagerPort) if err != nil { - return &exitError{code: 8, err: fmt.Errorf("setting up jobs-manager port-forward: %w", err)} + return false, &exitError{code: 8, err: fmt.Errorf("setting up jobs-manager port-forward: %w", err)} } defer pf.Close() @@ -810,7 +841,7 @@ other collaborators train against it without ever seeing the raw files.`)) // WatchResult Detached → 0 (cluster keeps running) // WatchResult Succeeded clean → 0 localEndpoint := fmt.Sprintf("http://localhost:%d", pf.LocalPort) - submitRes, err := submit.Run(ctx, submit.Options{ + submitRes, err := submitRunFn(ctx, submit.Options{ Submitter: submit.NewHTTPSubmitter(localEndpoint, tok.Token), Client: cs, IngestConfigYAML: string(specBytes), @@ -845,21 +876,13 @@ other collaborators train against it without ever seeing the raw files.`)) jsonEmitted = true } - // Reclaim the staged source copy on a CLEAN success only. The - // ingestor copies (not moves) the staged files into the table, so - // leaving .tracebloc-staging/ behind doubles PVC use for - // file-bearing datasets until the next --overwrite or `data delete` - // (the staging-leak found by the ingest UX audit; cli#166 / epic #67). - // Gated on status=="succeeded" so we never touch the source on a - // - detached run (status "detached"): the Job is still reading it; - // - partial (status "completed_with_failures") or failed run: the - // user may want the source to inspect/retry. - // Best-effort and time-bounded (push.StagingCleanupTimeout): a failed - // or slow reclaim must not fail — or noticeably delay — a successful - // ingest. - if status == "succeeded" { + // Reclaim the staged source copy on a CLEAN success only (see + // shouldReclaimStaging). Best-effort and time-bounded + // (push.StagingCleanupTimeout): a failed or slow reclaim must not + // fail — or noticeably delay — a successful ingest. + if shouldReclaimStaging(status) { a.Printer.Infof("Reclaiming the temporary staging copy on the cluster…") - if cerr := push.CleanStaging(ctx, cs, + if cerr := cleanStagingFn(ctx, cs, &push.SPDYExecutor{Config: resolved.RestConfig, Client: cs}, resolved.Namespace, a.Spec.Table, push.PodSpecOptions{ Namespace: resolved.Namespace, @@ -875,9 +898,25 @@ other collaborators train against it without ever seeing the raw files.`)) } if exitErr != nil { - return exitErr + return jsonEmitted, exitErr } - return nil + return jsonEmitted, nil +} + +// shouldReclaimStaging reports whether the staged source copy should be +// reclaimed after the run. ONLY on a clean success: the ingestor copies (not +// moves) the staged files into the table, so leaving .tracebloc-staging/
+// behind doubles PVC use for file-bearing datasets until the next --overwrite +// or `data delete` (the staging-leak found by the ingest UX audit; cli#166 / +// epic #67). Everything else keeps the source: +// - a detached run ("detached") — the Job is still reading it; +// - a partial ("completed_with_failures") or failed/errored run — the user +// may want the source to inspect or retry. +// +// This is the "must NOT reclaim on partial failure" gate (#1009), named so the +// invariant is table-testable in isolation. +func shouldReclaimStaging(status string) bool { + return status == "succeeded" } // classifyPushOutcome maps the result of submit.Run to a machine- @@ -1072,6 +1111,18 @@ func writePushErrorJSON(w io.Writer, sp push.SpecArgs, e error, code int) { // listDatasetsFn is a test seam over push.ListDatasets. var listDatasetsFn = push.ListDatasets +// Test seams over the cluster-touching steps of runIngestionRun (#1009). +// Production wires them to the real functions; a table test overrides them to +// drive the classify → exit-code → JSON → reclaim matrix without a cluster +// (mirrors the listDatasetsFn seam). cleanStagingFn is here too so a test can +// observe whether the staging reclaim ran (the must-NOT-reclaim gate). +var ( + mintIngestorTokenFn = cluster.MintIngestorToken + portForwardJobsManagerFn = submit.PortForwardJobsManager + submitRunFn = submit.Run + cleanStagingFn = push.CleanStaging +) + // destTableExists reports whether the destination table already holds an // ingested dataset, via the same query `data list` uses. It fails OPEN: a // broken check returns (false, note) so the ingest proceeds — the in-cluster diff --git a/internal/cli/data_test.go b/internal/cli/data_test.go index 703191e7..49716e1c 100644 --- a/internal/cli/data_test.go +++ b/internal/cli/data_test.go @@ -94,8 +94,8 @@ func execDataIngest(t *testing.T, args []string) (exitCode int, stdout, stderr s func TestDataIngest_UnsupportedCategory_ExitsTwo(t *testing.T) { root := imgcLayout(t) for _, badCategory := range []string{ - "semantic_segmentation", // blocked on the ingestor (data-ingestors#136) - "instance_segmentation", // not implemented + "semantic_segmentation", // known but blocked on the ingestor (data-ingestors#136) + "instance_segmentation", // dead — removed from the registry (#1005), now unrecognized "definitely-not-a-category", // nonsense; gate catches this too } { t.Run(badCategory, func(t *testing.T) { diff --git a/internal/cli/ingestion_run_test.go b/internal/cli/ingestion_run_test.go new file mode 100644 index 00000000..c7eb7af9 --- /dev/null +++ b/internal/cli/ingestion_run_test.go @@ -0,0 +1,170 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "testing" + + "github.com/tracebloc/cli/internal/cluster" + "github.com/tracebloc/cli/internal/push" + "github.com/tracebloc/cli/internal/submit" + "github.com/tracebloc/cli/internal/ui" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" +) + +// The money path (#1009): submit → classify → exit-code → JSON → reclaim. +// These tests pin the outcome matrix — including the "must NOT reclaim on +// partial failure" gate — without standing up a cluster, via the seams +// (mintIngestorTokenFn / portForwardJobsManagerFn / submitRunFn / +// cleanStagingFn) that runIngestionRun goes through. + +func succeededResult() *submit.Result { + return &submit.Result{ + Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "ingestor-x"}, + // A genuinely clean run: every stage counter equal (total == inserted + // == api_sent). APISentRecords must be set too — the ingestor-aligned + // HasFailures() treats api_sent < inserted as a partial, so omitting it + // (defaulting to 0) would misclassify this "succeeded" row as a failure. + Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 2, InsertedRecords: 2, APISentRecords: 2}}, + } +} + +func partialResult() *submit.Result { + return &submit.Result{ + Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "ingestor-x"}, + Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 2, InsertedRecords: 1, FailedRecords: 1}}, + } +} + +// TestShouldReclaimStaging pins the must-NOT-reclaim-on-partial gate: the +// staged source is reclaimed ONLY on a clean success. +func TestShouldReclaimStaging(t *testing.T) { + if !shouldReclaimStaging("succeeded") { + t.Error(`shouldReclaimStaging("succeeded") = false, want true`) + } + for _, st := range []string{ + "completed_with_failures", "failed", "unknown", "detached", + "auth_error", "submit_error", "watch_error", "dry-run", "error", "", + } { + if shouldReclaimStaging(st) { + t.Errorf("shouldReclaimStaging(%q) = true, want false — a non-clean run must keep the source", st) + } + } +} + +// TestRunIngestionRun_Matrix drives the whole outcome tail through the seams: +// per row it asserts the exit code, whether the staging reclaim ran, and the +// emitted --output-json status — all in lockstep. +func TestRunIngestionRun_Matrix(t *testing.T) { + origMint, origPF, origRun, origClean := mintIngestorTokenFn, portForwardJobsManagerFn, submitRunFn, cleanStagingFn + defer func() { + mintIngestorTokenFn, portForwardJobsManagerFn, submitRunFn, cleanStagingFn = origMint, origPF, origRun, origClean + }() + + target := &clusterTarget{ + Resolved: &cluster.ResolvedConfig{Namespace: "tracebloc"}, + Clientset: nil, // the seams ignore it; the reclaim SPDYExecutor literal doesn't deref + Release: &cluster.ParentRelease{IngestorSAName: "ingestor", JobsManagerServiceName: "jm", JobsManagerPort: 8080}, + PVC: &cluster.SharedPVC{ClaimName: "pvc", MountPath: "/data/shared"}, + } + spec := map[string]any{"table": "t", "category": "image_classification", "intent": "train", "label": "label"} + + cases := []struct { + name string + mintErr error + pfErr error + submitRes *submit.Result + submitErr error + wantCode int // 0 == success (nil err) + wantStatus string + wantReclaim bool + wantJSON bool + }{ + {"succeeded", nil, nil, succeededResult(), nil, 0, "succeeded", true, true}, + {"partial", nil, nil, partialResult(), nil, 9, "completed_with_failures", false, true}, + {"failed", nil, nil, &submit.Result{Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "j"}, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeFailed}}, nil, 9, "failed", false, true}, + {"detached", nil, nil, &submit.Result{Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "j"}, Watch: nil}, nil, 0, "detached", false, true}, + {"submit-auth", nil, nil, nil, &submit.SubmitError{StatusCode: 401}, 5, "auth_error", false, true}, + {"submit-5xx", nil, nil, nil, &submit.SubmitError{StatusCode: 500}, 8, "submit_error", false, true}, + {"watch-err", nil, nil, nil, &submit.WatchError{Err: errors.New("x")}, 9, "watch_error", false, true}, + // mint / port-forward failures return BEFORE the JSON emit and the + // reclaim; jsonEmitted is false (runDataIngest's error defer covers it). + {"mint-fail", errors.New("mint boom"), nil, nil, nil, 5, "", false, false}, + {"pf-fail", nil, errors.New("pf boom"), nil, nil, 8, "", false, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + reclaimCalled := false + mintIngestorTokenFn = func(_ context.Context, _ kubernetes.Interface, _, _ string, _ int64, _ []string) (*cluster.IngestorToken, error) { + if c.mintErr != nil { + return nil, c.mintErr + } + return &cluster.IngestorToken{Token: "tok"}, nil + } + portForwardJobsManagerFn = func(_ context.Context, _ kubernetes.Interface, _ *rest.Config, _, _ string, _ int) (*submit.ForwardedConnection, error) { + if c.pfErr != nil { + return nil, c.pfErr + } + return &submit.ForwardedConnection{LocalPort: 12345}, nil + } + submitRunFn = func(_ context.Context, _ submit.Options) (*submit.Result, error) { + return c.submitRes, c.submitErr + } + cleanStagingFn = func(_ context.Context, _ kubernetes.Interface, _ push.Executor, _, _ string, _ push.PodSpecOptions) error { + reclaimCalled = true + return nil + } + + var jsonBuf bytes.Buffer + a := runDataIngestArgs{ + Spec: push.SpecArgs{Table: "t"}, + Printer: ui.New(io.Discard, ui.WithColor(false)), + OutputJSON: true, + JSONOut: &jsonBuf, + } + je, err := runIngestionRun(context.Background(), io.Discard, a, target, []byte("yaml"), spec) + + code := 0 + if err != nil { + var ee *exitError + if !errors.As(err, &ee) { + t.Fatalf("err is not *exitError: %v", err) + } + code = ee.Code() + } + if code != c.wantCode { + t.Errorf("exit code = %d, want %d", code, c.wantCode) + } + if reclaimCalled != c.wantReclaim { + t.Errorf("reclaim called = %v, want %v (only a clean success reclaims)", reclaimCalled, c.wantReclaim) + } + if je != c.wantJSON { + t.Errorf("jsonEmitted = %v, want %v", je, c.wantJSON) + } + if c.wantJSON { + var got pushJSONResult + if err := json.Unmarshal(jsonBuf.Bytes(), &got); err != nil { + t.Fatalf("emitted JSON invalid: %v (%q)", err, jsonBuf.String()) + } + if got.Status != c.wantStatus { + t.Errorf("emitted JSON status = %q, want %q", got.Status, c.wantStatus) + } + } else if jsonBuf.Len() != 0 { + t.Errorf("expected no JSON on the pre-submit failure path, got %q", jsonBuf.String()) + } + }) + } +} + +// TestSeamsWiredToRealFns guards that the indirection didn't accidentally +// leave a seam nil (a nil seam would panic the money path in production). +func TestSeamsWiredToRealFns(t *testing.T) { + if mintIngestorTokenFn == nil || portForwardJobsManagerFn == nil || + submitRunFn == nil || cleanStagingFn == nil { + t.Fatal("a money-path seam is nil — production would panic") + } +} diff --git a/internal/cli/interactive.go b/internal/cli/interactive.go index 9ad71d01..d945de91 100644 --- a/internal/cli/interactive.go +++ b/internal/cli/interactive.go @@ -19,8 +19,8 @@ import ( // category picker. It derives from the push registry's CLI-supported // set — the exact categories runDataIngest's gate accepts — so the // picker can't drift from what `data ingest` actually supports. -// semantic_/instance_segmentation are excluded (CLISupported=false) -// until they're implemented. +// semantic_segmentation is excluded (CLISupported=false) until it's +// implemented. var promptCategories = push.SupportedCategoryIDs() // prompter is the narrow seam over the interactive library. Production diff --git a/internal/cluster/discover.go b/internal/cluster/discover.go index 00a7b7ff..f853e6da 100644 --- a/internal/cluster/discover.go +++ b/internal/cluster/discover.go @@ -297,15 +297,25 @@ const clientChartSelector = "app.kubernetes.io/name=client,app.kubernetes.io/man // carries the same CLIENT_ID under the same labels. // // This anchors R7 adopt-backfill: a live client whose backend cluster_id is null -// must be adopted (and its anchor backfilled), never re-minted. Best-effort — it -// returns (nil, nil) when nothing is installed or the cluster can't be read -// (unreachable / restricted RBAC), so callers fall back to a plain create. +// must be adopted (and its anchor backfilled), never re-minted. +// +// Return contract (deliberately three-valued, so callers can tell "empty" from +// "couldn't tell" — collapsing the two is what let `client create` mint a +// duplicate over a live client and orphan it, the phantom-1060 class): +// - (client, nil) — a live client was found; +// - (nil, nil) — the cluster is READABLE and genuinely has no client release; +// - (nil, err) — a read/RBAC error meant we could NOT determine either way. +// +// Callers must fail closed on the error case, never treat it as "nothing installed". func DiscoverInClusterClientID(ctx context.Context, cs kubernetes.Interface) (*InClusterClient, error) { deps, err := cs.AppsV1().Deployments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{ LabelSelector: clientChartSelector, }) if err != nil { - return nil, nil // best-effort: treat an unreadable cluster as "nothing installed" + // A reachable-but-unreadable cluster must NOT be reported as "nothing + // installed" — that ambiguity is exactly what let a duplicate be minted + // over a live client. Surface it so the caller fails closed. + return nil, fmt.Errorf("listing client deployments to check for an existing client: %w", err) } ns := "" for _, d := range deps.Items { @@ -315,20 +325,23 @@ func DiscoverInClusterClientID(ctx context.Context, cs kubernetes.Interface) (*I } } if ns == "" { - return nil, nil // no client release on this cluster + return nil, nil // readable, no client release installed — a genuine fresh cluster } secrets, err := cs.CoreV1().Secrets(ns).List(ctx, metav1.ListOptions{ LabelSelector: clientChartSelector, }) if err != nil { - return nil, nil + return nil, fmt.Errorf("reading the existing client's identity in namespace %q: %w", ns, err) } for _, s := range secrets.Items { if v, ok := s.Data["CLIENT_ID"]; ok && len(v) > 0 { return &InClusterClient{ClientID: string(v), Namespace: ns}, nil } } - return nil, nil + // A client release IS installed here (its jobs-manager Deployment exists) but its + // CLIENT_ID secret wasn't readable — we know a client is present, so this is not a + // fresh cluster. Fail closed rather than let the caller mint over it. + return nil, fmt.Errorf("a tracebloc client is installed in namespace %q but its CLIENT_ID could not be read", ns) } // pickJobsManagerService probes for the chart's jobs-manager diff --git a/internal/cluster/discover_test.go b/internal/cluster/discover_test.go index 10466c58..d316eb36 100644 --- a/internal/cluster/discover_test.go +++ b/internal/cluster/discover_test.go @@ -9,7 +9,9 @@ import ( appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" ) // jobsManagerDeployment builds the minimal Deployment the chart @@ -135,10 +137,42 @@ func TestDiscoverInClusterClientID_NoRelease(t *testing.T) { } func TestDiscoverInClusterClientID_ReleaseButNoSecret(t *testing.T) { + // A release IS installed (jobs-manager present) but its CLIENT_ID secret is + // absent/unreadable: we KNOW a client is here, so this must NOT read as "nothing + // installed". It returns an error so the caller fails closed rather than mint a + // duplicate over the live client (phantom-1060 class). cs := fake.NewClientset(jobsManagerDeployment("tracebloc", "tracebloc", "client-1.3.5", "1.3.5", "d")) got, err := DiscoverInClusterClientID(context.Background(), cs) - if err != nil || got != nil { - t.Errorf("release but no secret: want (nil,nil), got (%+v,%v)", got, err) + if err == nil || got != nil { + t.Errorf("release but unreadable CLIENT_ID: want (nil, error), got (%+v, %v)", got, err) + } +} + +func TestDiscoverInClusterClientID_DeploymentsListError_FailsClosed(t *testing.T) { + // A reachable-but-unreadable cluster (RBAC/transient List failure) must NOT be + // reported as (nil,nil) "nothing installed" — that ambiguity is what let a + // duplicate be minted over a live client. Surface an error so the caller fails + // closed. Regression guard for the phantom-1060 root cause. + cs := fake.NewClientset() + cs.PrependReactor("list", "deployments", func(k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, errors.New("forbidden: cannot list deployments") + }) + got, err := DiscoverInClusterClientID(context.Background(), cs) + if err == nil || got != nil { + t.Errorf("deployments list error: want (nil, error), got (%+v, %v)", got, err) + } +} + +func TestDiscoverInClusterClientID_SecretsListError_FailsClosed(t *testing.T) { + // A release is present but the secret read fails — still "couldn't determine", + // so return an error (fail closed), never (nil,nil). + cs := fake.NewClientset(jobsManagerDeployment("tracebloc", "tracebloc", "client-1.3.5", "1.3.5", "d")) + cs.PrependReactor("list", "secrets", func(k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, errors.New("forbidden: cannot list secrets") + }) + got, err := DiscoverInClusterClientID(context.Background(), cs) + if err == nil || got != nil { + t.Errorf("secrets list error: want (nil, error), got (%+v, %v)", got, err) } } diff --git a/internal/push/category.go b/internal/push/category.go index bfae858c..eef4a2e3 100644 --- a/internal/push/category.go +++ b/internal/push/category.go @@ -26,8 +26,8 @@ type CategorySpec struct { // never ships to the central backend by default. RegressionClass bool // CLISupported reports whether `dataset push` implements the category - // today. semantic_/instance_segmentation are known (the schema - // defines them) but not yet pushable. + // today. semantic_segmentation is known (the schema defines it) but + // not yet pushable. CLISupported bool // UnsupportedNote explains why a known-but-unimplemented category // isn't available yet; surfaced by the push gate. Empty when supported. @@ -49,11 +49,15 @@ const ( FamilyText ) -// categoryRegistry is the ordered, authoritative list of every category -// the ingest.v1 schema defines. Order is the display order for help text -// and the interactive picker (CLI-supported first, in workflow order; -// the not-yet-implemented ones last). Adding a category to the schema -// means adding it here — the parity test pins the set. +// categoryRegistry is the ordered list of every category the ingest.v1 +// schema defines — nothing more, nothing less. Order is the display order for +// help text and the interactive picker (CLI-supported first, in workflow +// order; the not-yet-implemented ones last). Adding a category to the schema +// means adding it here; TestRegistryCoversSchemaCategories + +// TestRegistryWithinSchema pin the set equal to the schema enum both ways, so +// it can neither fall behind (a schema category rejected as "unrecognized") +// nor carry an extra the ingestor won't accept (the instance_segmentation +// half-ingest class — data-ingestors #240/#99, #1005). var categoryRegistry = []CategorySpec{ {ID: "image_classification", Family: FamilyImage, Label: "Image classification", CLISupported: true}, {ID: "object_detection", Family: FamilyImage, Label: "Object detection", CLISupported: true}, @@ -66,8 +70,6 @@ var categoryRegistry = []CategorySpec{ {ID: "time_to_event_prediction", Family: FamilyTabular, Label: "Time-to-event prediction", RegressionClass: true, CLISupported: true}, {ID: "semantic_segmentation", Family: FamilyImage, Label: "Semantic segmentation", CLISupported: false, UnsupportedNote: "blocked on the ingestor's mask-sidecar support (data-ingestors#136)"}, - {ID: "instance_segmentation", Family: FamilyImage, Label: "Instance segmentation", CLISupported: false, - UnsupportedNote: "not implemented"}, {ID: "causal_language_modeling", Family: FamilyText, Label: "Causal language modeling", CLISupported: false, UnsupportedNote: "schema-recognized (data-ingestors#805); `tracebloc ingest` discover/build for its raw-.txt / prompt\\tcompletion `texts` layout is pending"}, {ID: "seq2seq", Family: FamilyText, Label: "Sequence-to-sequence", CLISupported: false, diff --git a/internal/push/category_registry_test.go b/internal/push/category_registry_test.go index eb38a406..3548a381 100644 --- a/internal/push/category_registry_test.go +++ b/internal/push/category_registry_test.go @@ -15,7 +15,7 @@ import ( func TestRegistryKnownCategories(t *testing.T) { want := []string{ "image_classification", "object_detection", "keypoint_detection", - "semantic_segmentation", "instance_segmentation", + "semantic_segmentation", "text_classification", "token_classification", "masked_language_modeling", "causal_language_modeling", "seq2seq", "sentence_pair_classification", "embeddings", @@ -45,9 +45,9 @@ func TestSupportedCategories(t *testing.T) { t.Errorf("SupportedCategoryIDs returned %q but IsCLISupported is false", id) } } - // segmentation + the self-supervised text categories (CLM, seq2seq) + - // token_classification are known but not yet pushable, and must explain why. - for _, id := range []string{"semantic_segmentation", "instance_segmentation", "causal_language_modeling", "seq2seq", "token_classification", "sentence_pair_classification", "embeddings"} { + // semantic_segmentation + the self-supervised text categories (CLM, seq2seq) + // + token_classification are known but not yet pushable, and must explain why. + for _, id := range []string{"semantic_segmentation", "causal_language_modeling", "seq2seq", "token_classification", "sentence_pair_classification", "embeddings"} { if !IsKnown(id) { t.Errorf("%s should be known", id) } @@ -87,16 +87,12 @@ func TestPredicatesDeriveFromRegistry(t *testing.T) { } } -// TestRegistryCoversSchemaCategories pins registry⇄schema parity: every -// category the ingest schema accepts must be known to the registry, or a -// schema-valid `dataset push --category=X` is wrongly rejected as -// "unrecognized" (the token_classification drift, Bugbot v0.4.0 RC). The -// existing tests only pin the registry against a hand-written list, which -// stays internally consistent while drifting from the schema — this closes -// that gap. The reverse direction isn't required: the registry may carry a -// known-but-unsupported alias the v1 schema doesn't list yet (e.g. -// instance_segmentation), which is gated out before schema validation. -func TestRegistryCoversSchemaCategories(t *testing.T) { +// schemaCategoryEnum returns the category enum from the embedded ingest.v1 +// schema — the single source of truth the registry is pinned against (#1005). +// The schema is vendored + drift-checked against data-ingestors by +// scripts/sync-schema.sh, so this ties the registry transitively to upstream. +func schemaCategoryEnum(t *testing.T) []string { + t.Helper() var doc struct { Properties struct { Category struct { @@ -110,7 +106,23 @@ func TestRegistryCoversSchemaCategories(t *testing.T) { if len(doc.Properties.Category.Enum) == 0 { t.Fatal("no category enum found in the embedded schema (parse path wrong?)") } - for _, id := range doc.Properties.Category.Enum { + return doc.Properties.Category.Enum +} + +// registryAliases are registry category IDs deliberately NOT in the ingest.v1 +// schema enum — declared placeholders. Empty today: instance_segmentation used +// to sit here unchecked, but it's dead (it half-ingested with no validators or +// file transfer — data-ingestors #240/#99) and was removed, not kept. A future +// known-but-unschema'd placeholder must be DECLARED here, so TestRegistryWithinSchema +// flags undeclared drift while allowing an intentional superset (#1005). +var registryAliases = map[string]bool{} + +// TestRegistryCoversSchemaCategories pins schema ⊆ registry: every category the +// ingest schema accepts must be known to the registry, or a schema-valid +// `dataset push --category=X` is wrongly rejected as "unrecognized" (the +// token_classification drift, Bugbot v0.4.0 RC). +func TestRegistryCoversSchemaCategories(t *testing.T) { + for _, id := range schemaCategoryEnum(t) { if !IsKnown(id) { t.Errorf("schema category %q missing from the registry — `dataset push --category=%s` "+ "would be rejected as unrecognized despite passing schema validation", id, id) @@ -118,6 +130,28 @@ func TestRegistryCoversSchemaCategories(t *testing.T) { } } +// TestRegistryWithinSchema pins registry ⊆ schema (+ declared aliases): the +// registry must not carry a category the ingest schema — and therefore the +// ingestor — doesn't accept. An undeclared extra is exactly the +// instance_segmentation half-ingest class: the backend/CLI would accept a +// `--category` the pipeline can't handle, and the config half-ingests (DB rows +// + API records, zero files staged; #1005, data-ingestors #240/#99). Together +// with TestRegistryCoversSchemaCategories this pins registry == schema, modulo +// explicitly declared placeholders in registryAliases. +func TestRegistryWithinSchema(t *testing.T) { + inSchema := make(map[string]bool) + for _, id := range schemaCategoryEnum(t) { + inSchema[id] = true + } + for _, id := range AllCategoryIDs() { + if !inSchema[id] && !registryAliases[id] { + t.Errorf("registry category %q is not in the ingest.v1 schema enum and not a declared "+ + "alias — add it to the schema (data-ingestors) if it's real, or declare it in "+ + "registryAliases if it's an intentional placeholder", id) + } + } +} + func equalSet(a, b []string) bool { if len(a) != len(b) { return false diff --git a/internal/push/parity_golden_test.go b/internal/push/parity_golden_test.go index 2006b39d..186db758 100644 --- a/internal/push/parity_golden_test.go +++ b/internal/push/parity_golden_test.go @@ -4,11 +4,13 @@ import ( "encoding/json" "os" "path/filepath" + "slices" "sort" "testing" ) -// The validator-parity harness (backend#828 P3). Two assertions per case: +// The validator-parity harness (backend#828 P3; value-level from backend#1009). +// Per case: // // 1. the Go preflight's verdict matches the manifest's cli_verdict — // pins the CLI side; @@ -17,6 +19,11 @@ import ( // manifest's ingestor_verdict — so when the ingestor's rules change, // regenerating the goldens fails this test until the manifest (and, // where needed, the Go preview) is consciously updated. +// 3. for cases flagged value_parity, the Go preview's VALUE-level read of +// the label column (resolved header + row count + class set) equals the +// REAL ingestor's — the only assertion that catches accept/accept with +// divergent stored data (data-ingestors #340: a case-/whitespace- +// mismatched label passes both verdicts, then reads null in-cluster). // // Deliberate divergences (the CLI previewing read-/transfer-time failures // the ingestor's preflight can't see) are explicit in the manifest, never @@ -32,6 +39,7 @@ type parityCase struct { Schema map[string]string `json:"schema"` CLIVerdict string `json:"cli_verdict"` IngestorVerdict string `json:"ingestor_verdict"` + ValueParity bool `json:"value_parity"` Note string `json:"note"` } @@ -45,6 +53,11 @@ func TestValidatorParity(t *testing.T) { Verdicts map[string]struct { Verdict string `json:"verdict"` Errors []string `json:"errors"` + Values *struct { + Resolved string `json:"resolved_label"` + RowCount int `json:"row_count"` + Classes []string `json:"classes"` + } `json:"values"` } `json:"verdicts"` } mustLoad(t, filepath.Join("testdata", "parity", "goldens.json"), &goldens) @@ -64,10 +77,56 @@ func TestValidatorParity(t *testing.T) { if got != c.CLIVerdict { t.Errorf("Go preflight = %q, manifest expects %q (note: %s)", got, c.CLIVerdict, c.Note) } + + if !c.ValueParity { + return + } + // Value-level parity (backend#1009): the Go preview must read the + // SAME label header, row count, and class set the real ingestor + // does. Catches accept/accept-with-divergent-label (#340). + if golden.Values == nil { + t.Fatalf("case %s is value_parity but goldens.json has no values — "+ + "regenerate with scripts/gen-validator-goldens.py against a data-ingestors "+ + "checkout that includes the #340 label-resolution fix", c.Name) + } + gv := goLabelValues(t, c) + if gv.Resolved != golden.Values.Resolved { + t.Errorf("resolved label: Go preview = %q, ingestor golden = %q "+ + "(the read paths resolve the label column differently — #340 class)", + gv.Resolved, golden.Values.Resolved) + } + if gv.RowCount != golden.Values.RowCount { + t.Errorf("row count: Go preview = %d, ingestor golden = %d", gv.RowCount, golden.Values.RowCount) + } + if !slices.Equal(gv.Classes, golden.Values.Classes) { + t.Errorf("class set: Go preview = %v, ingestor golden = %v", gv.Classes, golden.Values.Classes) + } }) } } +// goLabelValues runs the Go preview's value-level label read for a case, +// deriving the NA-drop / numeric-collapse flags from the label's schema type +// exactly as PreflightDataset does — so the value comparison uses the same +// read semantics the production preflight would. +func goLabelValues(t *testing.T, c parityCase) LabelReadValues { + t.Helper() + csvPath := filepath.Join("testdata", "parity", "cases", c.Name, c.CSV) + schema := c.Schema + if IsTabular(c.Category) && len(schema) == 0 { + if sch, _, _, err := InferSchema(csvPath); err == nil { + schema = sch + } + } + dropNA, collapse := false, false + if IsTabular(c.Category) { + sqlType, inSchema := labelSchemaType(schema, c.LabelColumn) + dropNA = inSchema + collapse = !(inSchema && isStringSQLType(sqlType)) + } + return ReadLabelValues(csvPath, c.LabelColumn, dropNA, collapse) +} + // runGoPreflight runs THE production dispatch (push.PreflightDataset) over // the case — the same code path runDataIngest executes, so a check deleted // or rewired in production fails parity here. diff --git a/internal/push/preflight.go b/internal/push/preflight.go index 0e67e7e5..a2810eaf 100644 --- a/internal/push/preflight.go +++ b/internal/push/preflight.go @@ -383,9 +383,54 @@ func TruncateList(items []string, max int) string { // even an empty string is a real class and every distinct trimmed string // counts. The caller derives the two flags from the label's schema type. func CheckLabelDiversity(csvPath, labelColumn string, dropNASentinels, collapseNumeric bool) error { + v := readLabelColumnValues(csvPath, labelColumn, dropNASentinels, collapseNumeric) + // Benign-skip when the column is absent (that's CheckLabelColumn's + // diagnostic) or an unreadable file (another check's) — both leave Found + // false. Two or more classes is diverse enough. + if !v.Found || len(v.Classes) >= 2 { + return nil + } + return fmt.Errorf( + "the label column %q has %d distinct value(s) — a classification dataset needs at "+ + "least 2 classes. The cluster rejects this after the upload; check the labels and re-run.", + labelColumn, len(v.Classes)) +} + +// LabelReadValues is the value-level view of a label column: the header the +// read path RESOLVES the configured name to (case/whitespace-insensitively — +// the ingestor's rule), the sorted distinct classes the ingestor counts, and +// the data-row count. It is what the value-level parity harness pins, so a +// preview that says "N rows, K classes" cannot silently diverge from what the +// ingestor actually reads — the accept/accept-with-divergent-label class the +// verdict-only harness is blind to (data-ingestors #340). +type LabelReadValues struct { + Resolved string `json:"resolved_label"` + Classes []string `json:"classes"` + RowCount int `json:"row_count"` + Found bool `json:"-"` +} + +// ReadLabelValues is the exported value-level read used by the parity harness +// (and, later, the RFC-0002 "check your data" preview). It shares the exact +// read/resolve/NA/collapse rules with CheckLabelDiversity via +// readLabelColumnValues, so the value preview and the diversity verdict cannot +// drift from each other. +func ReadLabelValues(csvPath, labelColumn string, dropNASentinels, collapseNumeric bool) LabelReadValues { + return readLabelColumnValues(csvPath, labelColumn, dropNASentinels, collapseNumeric) +} + +// readLabelColumnValues reads csvPath's label column once and returns its +// value-level view. The column is resolved exactly, then case/whitespace- +// insensitively (mirroring the ingestor's resolve_column rule); each row value +// is whitespace-trimmed; NA sentinels are dropped and numeric values collapsed +// per the caller's flags (see CheckLabelDiversity's doc for how those mirror +// the ingestor's per-column read). Unlike the previous early-exit diversity +// scan, this reads the whole column to build the full class set + row count — +// one scan now backs both the diversity verdict and the value-level preview. +func readLabelColumnValues(csvPath, labelColumn string, dropNASentinels, collapseNumeric bool) LabelReadValues { f, err := os.Open(csvPath) if err != nil { - return nil // unreadable file is another check's diagnostic + return LabelReadValues{} // Found=false: unreadable file is another check's diagnostic } defer func() { _ = f.Close() }() br := bufio.NewReader(f) @@ -396,12 +441,12 @@ func CheckLabelDiversity(csvPath, labelColumn string, dropNASentinels, collapseN r.FieldsPerRecord = -1 header, err := r.Read() if err != nil { - return nil + return LabelReadValues{} } - col := -1 + col, resolved := -1, "" for i, c := range header { if strings.TrimSpace(c) == labelColumn { - col = i + col, resolved = i, strings.TrimSpace(c) break } } @@ -409,21 +454,26 @@ func CheckLabelDiversity(csvPath, labelColumn string, dropNASentinels, collapseN want := strings.ToLower(strings.TrimSpace(labelColumn)) for i, c := range header { if strings.ToLower(strings.TrimSpace(c)) == want { - col = i + col, resolved = i, strings.TrimSpace(c) break } } } if col == -1 { - return nil // benign-skip, like the ingestor + return LabelReadValues{} // Found=false — benign skip, like the ingestor } distinct := map[string]bool{} + rowCount := 0 for { rec, err := r.Read() if errors.Is(err, io.EOF) { break } - if err != nil || len(rec) <= col { + if err != nil { + continue + } + rowCount++ + if len(rec) <= col { continue } v := strings.TrimSpace(rec[col]) @@ -435,22 +485,18 @@ func CheckLabelDiversity(csvPath, labelColumn string, dropNASentinels, collapseN if collapseNumeric { // Numeric inference collapses "1" and "1.0" into one value // in-cluster; normalize the same way before counting. - if f, err := strconv.ParseFloat(v, 64); err == nil { - v = strconv.FormatFloat(f, 'g', -1, 64) + if fv, err := strconv.ParseFloat(v, 64); err == nil { + v = strconv.FormatFloat(fv, 'g', -1, 64) } } distinct[v] = true - if len(distinct) >= 2 { - return nil - } } - if len(distinct) >= 2 { - return nil + classes := make([]string, 0, len(distinct)) + for k := range distinct { + classes = append(classes, k) } - return fmt.Errorf( - "the label column %q has %d distinct value(s) — a classification dataset needs at "+ - "least 2 classes. The cluster rejects this after the upload; check the labels and re-run.", - labelColumn, len(distinct)) + sort.Strings(classes) + return LabelReadValues{Resolved: resolved, Classes: classes, RowCount: rowCount, Found: true} } // knownMediaExtensions mirrors the ingestor's FileExtension.get_all_extensions diff --git a/internal/push/testdata/parity/cases.json b/internal/push/testdata/parity/cases.json index 27ff423a..b54b8729 100644 --- a/internal/push/testdata/parity/cases.json +++ b/internal/push/testdata/parity/cases.json @@ -7,7 +7,8 @@ "csv": "data.csv", "label_column": "label", "cli_verdict": "accept", - "ingestor_verdict": "accept" + "ingestor_verdict": "accept", + "value_parity": true }, { "name": "tabular-dup-header", @@ -56,7 +57,8 @@ 8 ], "cli_verdict": "accept", - "ingestor_verdict": "accept" + "ingestor_verdict": "accept", + "value_parity": true }, { "name": "imgc-bom-labels", @@ -70,7 +72,8 @@ ], "cli_verdict": "accept", "ingestor_verdict": "accept", - "note": "pandas strips the BOM in-cluster \u2014 the CLI must NOT reject what the cluster accepts (cli#71 parity)" + "note": "pandas strips the BOM in-cluster \u2014 the CLI must NOT reject what the cluster accepts (cli#71 parity)", + "value_parity": true }, { "name": "imgc-label-missing", @@ -98,7 +101,8 @@ ], "cli_verdict": "accept", "ingestor_verdict": "accept", - "note": "the ingestor's _match_column is case-insensitive + trimmed \u2014 the CLI must be exactly as loose" + "note": "the ingestor's _match_column is case-insensitive + trimmed \u2014 the CLI must be exactly as loose", + "value_parity": true }, { "name": "imgc-zero-byte", @@ -196,7 +200,8 @@ ], "cli_verdict": "accept", "ingestor_verdict": "accept", - "note": "non-square [W,H]=[8,4]: pins the target_size ORIENTATION end-to-end \u2014 an [H,W] swap on emit (the pre-P3 bug) flips this to reject in-cluster" + "note": "non-square [W,H]=[8,4]: pins the target_size ORIENTATION end-to-end \u2014 an [H,W] swap on emit (the pre-P3 bug) flips this to reject in-cluster", + "value_parity": true }, { "name": "imgc-nonsquare-swapped", @@ -238,7 +243,8 @@ ], "cli_verdict": "accept", "ingestor_verdict": "accept", - "note": "empty-string label IS a class for image categories (keep_default_na=False): the CLI must not be stricter" + "note": "empty-string label IS a class for image categories (keep_default_na=False): the CLI must not be stricter", + "value_parity": true }, { "name": "imgc-dotted-stem", @@ -252,7 +258,8 @@ ], "cli_verdict": "accept", "ingestor_verdict": "accept", - "note": "row 'photo.2024' resolves to photo.2024.jpg via _has_extension (dotted stems are not extensions) \u2014 the CLI's cross-check must mirror, not reject" + "note": "row 'photo.2024' resolves to photo.2024.jpg via _has_extension (dotted stems are not extensions) \u2014 the CLI's cross-check must mirror, not reject", + "value_parity": true }, { "name": "tabular-na-labels", @@ -280,7 +287,8 @@ "extension": ".txt", "cli_verdict": "accept", "ingestor_verdict": "accept", - "note": "pins the text-family dispatch" + "note": "pins the text-family dispatch", + "value_parity": true }, { "name": "tabular-varchar-numeric-labels", @@ -293,7 +301,8 @@ }, "cli_verdict": "accept", "ingestor_verdict": "accept", - "note": "labels '1' vs '1.0' under a VARCHAR label: the ingestor pins dtype=str (no numeric collapse) so they are 2 classes \u2014 pins #152's schema-type-aware collapse (the earlier blanket collapse falsely rejected this)" + "note": "labels '1' vs '1.0' under a VARCHAR label: the ingestor pins dtype=str (no numeric collapse) so they are 2 classes \u2014 pins #152's schema-type-aware collapse (the earlier blanket collapse falsely rejected this)", + "value_parity": true }, { "name": "tabular-float-numeric-labels", diff --git a/internal/push/testdata/parity/goldens.json b/internal/push/testdata/parity/goldens.json index ac7553dd..c67d17e7 100644 --- a/internal/push/testdata/parity/goldens.json +++ b/internal/push/testdata/parity/goldens.json @@ -3,6 +3,14 @@ "verdicts": { "imgc-bom-labels": { "errors": [], + "values": { + "classes": [ + "cat", + "dog" + ], + "resolved_label": "label", + "row_count": 2 + }, "verdict": "accept" }, "imgc-corrupt": { @@ -13,6 +21,14 @@ }, "imgc-dotted-stem": { "errors": [], + "values": { + "classes": [ + "cat", + "dog" + ], + "resolved_label": "label", + "row_count": 2 + }, "verdict": "accept" }, "imgc-dup-header": { @@ -21,6 +37,14 @@ }, "imgc-empty-label": { "errors": [], + "values": { + "classes": [ + "", + "A" + ], + "resolved_label": "label", + "row_count": 2 + }, "verdict": "accept" }, "imgc-header-only": { @@ -31,6 +55,14 @@ }, "imgc-label-case": { "errors": [], + "values": { + "classes": [ + "cat", + "dog" + ], + "resolved_label": "Label", + "row_count": 2 + }, "verdict": "accept" }, "imgc-label-missing": { @@ -51,6 +83,14 @@ }, "imgc-nonsquare": { "errors": [], + "values": { + "classes": [ + "cat", + "dog" + ], + "resolved_label": "label", + "row_count": 2 + }, "verdict": "accept" }, "imgc-nonsquare-swapped": { @@ -61,6 +101,14 @@ }, "imgc-ok": { "errors": [], + "values": { + "classes": [ + "cat", + "dog" + ], + "resolved_label": "label", + "row_count": 2 + }, "verdict": "accept" }, "imgc-res-mismatch": { @@ -123,14 +171,38 @@ }, "tabular-ok": { "errors": [], + "values": { + "classes": [ + "0", + "1" + ], + "resolved_label": "label", + "row_count": 2 + }, "verdict": "accept" }, "tabular-varchar-numeric-labels": { "errors": [], + "values": { + "classes": [ + "1", + "1.0" + ], + "resolved_label": "label", + "row_count": 2 + }, "verdict": "accept" }, "text-clf-ok": { "errors": [], + "values": { + "classes": [ + "neg", + "pos" + ], + "resolved_label": "label", + "row_count": 2 + }, "verdict": "accept" } } diff --git a/internal/submit/portforward.go b/internal/submit/portforward.go index 20c6155a..1dc2cefd 100644 --- a/internal/submit/portforward.go +++ b/internal/submit/portforward.go @@ -30,8 +30,13 @@ type ForwardedConnection struct { done chan struct{} } -// Close tears down the port-forward. Safe to call multiple times. +// Close tears down the port-forward. Safe to call multiple times, and +// safe on a zero-value connection that was never started (stopCh nil) — +// e.g. a test fake handed back by an injected PortForwardJobsManager. func (f *ForwardedConnection) Close() { + if f.stopCh == nil { + return // never started; nothing to tear down + } select { case <-f.stopCh: return // already closed diff --git a/internal/submit/summary.go b/internal/submit/summary.go index eb0a40b7..0f19042c 100644 --- a/internal/submit/summary.go +++ b/internal/submit/summary.go @@ -69,26 +69,41 @@ type Summary struct { FailedRecords int64 } -// HasFailures returns true if any failure-class counter is non-zero. -// Used by the orchestrator to decide which exit code to return -// (success: 0, ingest-with-failures: non-zero) and how to color -// the rendered panel. +// HasFailures returns true if any non-trivial failure occurred. It MIRRORS +// the ingestor's IngestionSummary.has_failures EXACTLY (data-ingestors +// ingestors/base.py) — DB insert short of total, API short of inserted, a +// file-transfer or processing drop (skipped), or a hard failure — so the CLI's +// exit code + staging-reclaim gate agree with the ingestor's own "completed +// successfully" banner. The narrower prior version (only FailedRecords / +// FileTransferFailures) reported success and reclaimed the staged source on a +// run that silently SKIPPED rows or inserted fewer than total — silent data +// loss that then deletes the user's only copy. Every counter this reads is +// emitted unconditionally by the ingestor banner + parsed above, so the +// inserted 0 || s.FailedRecords > 0 + return s.FailedRecords > 0 || + s.FileTransferFailures > 0 || + s.SkippedRecords > 0 || + s.InsertedRecords < s.TotalRecords || + s.APISentRecords < s.InsertedRecords } -// SuccessRate returns a 0-100 percentage for the panel header. -// Defined as ProcessedRecords / TotalRecords; returns 0 when -// TotalRecords is 0 to avoid divide-by-zero in early-failure -// banners. +// SuccessRate returns a 0-100 percentage for the panel header. Defined as +// InsertedRecords / TotalRecords — matching the ingestor's own banner +// (reporting.py: inserted_records / total_records), since InsertedRecords (rows +// that actually landed in MySQL) is the metric that matters for training, and +// ProcessedRecords (passed validation) is a superset that OVERSTATED success +// when rows validated but failed to insert. Returns 0 when TotalRecords is 0 to +// avoid divide-by-zero in early-failure banners. func (s *Summary) SuccessRate() float64 { if s == nil || s.TotalRecords == 0 { return 0 } - return float64(s.ProcessedRecords) / float64(s.TotalRecords) * 100 + return float64(s.InsertedRecords) / float64(s.TotalRecords) * 100 } // ansiCodeRE matches the ANSI SGR (Select Graphic Rendition) @@ -337,10 +352,21 @@ func RenderSummary(p *ui.Printer, s *Summary) { headline := fmt.Sprintf("ingested %s of %s records (%.1f%%)", commaSep(s.InsertedRecords), commaSep(s.TotalRecords), s.SuccessRate()) switch { - case s.HasFailures(): + case s.FailedRecords > 0 || s.FileTransferFailures > 0: + // Hard failures: rows errored at DB insert or file transfer. p.Errorf("Ingestion completed with failures — %s", headline) - case s.SkippedRecords > 0: - p.Warnf("Ingestion completed with skips — %s", headline) + case s.HasFailures(): + // No hard failure, but not clean: rows skipped, or fewer inserted/ + // synced than the ingestor saw. Exit-coded as not-clean (HasFailures), + // but colored distinctly from a hard failure. Word it by which soft + // shortfall actually occurred — "skips" only when rows were skipped; + // an insert/API shortfall with zero skips is a partial result, not a + // skip, and mislabeling it reads as a validator drop. + if s.SkippedRecords > 0 { + p.Warnf("Ingestion completed with skips — %s", headline) + } else { + p.Warnf("Ingestion completed partially — %s", headline) + } default: p.Successf("Ingestion complete — %s", headline) } diff --git a/internal/submit/summary_test.go b/internal/submit/summary_test.go index 6df9546c..58ab6b73 100644 --- a/internal/submit/summary_test.go +++ b/internal/submit/summary_test.go @@ -71,20 +71,24 @@ func TestSummaryParser_RealBannerEndToEnd(t *testing.T) { // that the orchestrator uses to choose between success exit code // (0) and ingest-failure exit code (9). func TestSummaryParser_HasFailures(t *testing.T) { + // Mirrors the ingestor's IngestionSummary.has_failures exactly. cases := []struct { name string s *Summary want bool }{ {"nil", nil, false}, - {"all zero", &Summary{TotalRecords: 100, ProcessedRecords: 100}, false}, - {"file transfer failures", &Summary{FileTransferFailures: 1}, true}, - {"failed records", &Summary{FailedRecords: 1}, true}, - {"both", &Summary{FileTransferFailures: 1, FailedRecords: 1}, true}, - // Skipped records are NOT failures — they're rows that - // validators rejected. The customer wants to see the - // count but it doesn't change the exit code. - {"skipped is not failure", &Summary{SkippedRecords: 100}, false}, + // A genuinely clean run: every counter equal, nothing skipped/failed. + {"clean", &Summary{TotalRecords: 100, ProcessedRecords: 100, InsertedRecords: 100, APISentRecords: 100}, false}, + {"file transfer failures", &Summary{TotalRecords: 1, InsertedRecords: 1, APISentRecords: 1, FileTransferFailures: 1}, true}, + {"failed records", &Summary{TotalRecords: 1, InsertedRecords: 1, APISentRecords: 1, FailedRecords: 1}, true}, + // Skipped rows ARE a failure — a dropped row is silent data loss + // (#234); the ingestor counts it, so the CLI must too (was the bug). + {"skipped is a failure", &Summary{TotalRecords: 100, InsertedRecords: 100, APISentRecords: 100, SkippedRecords: 5}, true}, + // Fewer rows in MySQL than the ingestor saw → partial run. + {"inserted < total", &Summary{TotalRecords: 100, ProcessedRecords: 100, InsertedRecords: 99, APISentRecords: 99}, true}, + // Rows in MySQL but the central catalog got fewer. + {"api_sent < inserted", &Summary{TotalRecords: 100, InsertedRecords: 100, APISentRecords: 99}, true}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { @@ -99,6 +103,7 @@ func TestSummaryParser_HasFailures(t *testing.T) { // rendered panel's "Success rate: XX%" line. Divide-by-zero on // empty banner is the critical edge case. func TestSummaryParser_SuccessRate(t *testing.T) { + // Rate is INSERTED/total (matches the ingestor banner), not processed/total. cases := []struct { name string s *Summary @@ -106,8 +111,11 @@ func TestSummaryParser_SuccessRate(t *testing.T) { }{ {"nil", nil, 0}, {"empty banner", &Summary{}, 0}, - {"100%", &Summary{TotalRecords: 100, ProcessedRecords: 100}, 100}, - {"50%", &Summary{TotalRecords: 100, ProcessedRecords: 50}, 50}, + {"100%", &Summary{TotalRecords: 100, ProcessedRecords: 100, InsertedRecords: 100}, 100}, + {"50%", &Summary{TotalRecords: 100, InsertedRecords: 50}, 50}, + // The overstatement the fix closes: all rows validated (processed=100) + // but only 70 landed in MySQL → 70%, not the old 100%. + {"processed overstates: inserted&2 + status=1 + continue + fi + line="$(go test -cover "./$pkg/" 2>/dev/null | grep -E 'coverage: [0-9]' || true)" + pct="$(printf '%s\n' "$line" | sed -nE 's/.*coverage: ([0-9]+(\.[0-9]+)?)% of statements.*/\1/p' | head -1)" + if [ -z "$pct" ]; then + echo "::error::could not read coverage for ./$pkg/ (did any test run?)" >&2 + status=1 + continue + fi + # awk exits 0 when pct < min (i.e. below the floor → failure). + if awk "BEGIN{exit !($pct < $min)}"; then + echo "::error::./$pkg/ coverage ${pct}% is below the floor ${min}% — add tests, or (with a reason) lower the floor in scripts/coverage-floor.sh" >&2 + status=1 + else + echo "ok: ./$pkg/ ${pct}% >= ${min}%" + fi +done + +exit "$status" diff --git a/scripts/gen-validator-goldens.py b/scripts/gen-validator-goldens.py index f87aebb5..99ed3b20 100644 --- a/scripts/gen-validator-goldens.py +++ b/scripts/gen-validator-goldens.py @@ -22,6 +22,16 @@ DuplicateValidator are skipped (they check cluster-side state — the table name is validated separately by both sides, and destination-duplicate handling is the cli#70 guard's territory, not a data-hygiene rule). + +For cases the manifest flags ``value_parity`` (and the ingestor accepts), +this also records a VALUE-level golden — the label column the ingestor read +path RESOLVES to, the row count, and the class set it stores — by driving the +REAL read path (CSVIngestor.read_data + the #340 label resolution + +RecordProcessor). parity_golden_test.go then pins that the Go preview reads +exactly the same values, catching accept/accept-with-divergent-label — the +#340 class a verdict alone is blind to (backend#1009). This requires the #340 +fix in the target ingestor; the generator fails loudly without it rather than +pin the bug. """ import json @@ -60,6 +70,59 @@ def infer_schema(csv_path): return {str(c).strip(): "VARCHAR(255)" for c in cols} +def read_label_values(case, csv_path, cfg, options): + """Drive the REAL ingestor read path — CSVIngestor.read_data + the #340 + label-column resolution + RecordProcessor — to capture the value-level view + the parity harness pins: the resolved label header, the row count, and the + sorted distinct classes the ingestor actually stores. This is the only + thing that catches accept/accept-with-divergent-label (the #340 class): + verdicts stay 'accept' while the stored labels silently go null. + + Requires the #340 fix (BaseIngestor._resolve_label_column) in the target + ingestor — without it a case-/whitespace-mismatched label would read null + and this generator would pin the BUG. Fails loudly if it's absent. + """ + from unittest.mock import MagicMock + + from tracebloc_ingestor.ingestors.csv_ingestor import CSVIngestor + + db = MagicMock() + db.config = cfg + file_opts = {k: v for k, v in options.items() if k != "schema"} + ing = CSVIngestor( + database=db, + api_client=MagicMock(), + table_name="parity_t", + schema=options.get("schema", {}) or {}, + label_column=case.get("label_column", "label"), + intent="train", + category=case["category"], + file_options=file_opts, + ) + if not hasattr(ing, "_resolve_label_column"): + sys.exit( + "the target ingestor predates the #340 label-resolution fix; " + "value-level parity requires it. Point DATA_INGESTORS_DIR at a " + "checkout that includes BaseIngestor._resolve_label_column." + ) + records = list(ing.read_data(csv_path)) + # Pin the label column on the first record that CONTAINS it (mirrors the + # ingest loop; sparse-record-safe), then read every row's stored label. + for rec in records: + if ing._resolve_label_column(rec.keys()): + break + labels = [] + for rec in records: + cleaned = ing.process_record(rec) + labels.append(cleaned.get("label") if cleaned else None) + classes = sorted({str(v) for v in labels if v is not None}) + return { + "resolved_label": ing.label_column, + "row_count": len(records), + "classes": classes, + } + + def run_case(case): case_dir = os.path.join(PARITY, "cases", case["name"]) csv_path = os.path.join(case_dir, case["csv"]) @@ -103,10 +166,18 @@ def run_case(case): except Exception as exc: # a raising validator is a rejection too errors.append(f"{type(v).__name__}: raised {exc}") - return { + result = { "verdict": "reject" if errors else "accept", "errors": errors[:6], } + # Value-level golden (data-ingestors #340 class): for cases the manifest + # flags value_parity AND the ingestor accepts, pin the resolved label, + # row count, and class set the REAL read path produces — parity_golden_test + # asserts the Go preview reads exactly these. Only meaningful when accepted + # (a rejected run never reaches the read path). + if case.get("value_parity") and not errors: + result["values"] = read_label_values(case, csv_path, cfg, options) + return result def main(): diff --git a/scripts/sync-schema.sh b/scripts/sync-schema.sh index a7f13d48..d4b42a4c 100755 --- a/scripts/sync-schema.sh +++ b/scripts/sync-schema.sh @@ -16,8 +16,15 @@ # scripts/sync-schema.sh --check # verify in-tree copy matches upstream; exit non-zero on drift # # Env knobs: -# SCHEMA_SOURCE_URL override the upstream URL (default: data-ingestors' master) -# SCHEMA_OUT override the in-tree destination (default: internal/schema/ingest.v1.json) +# SCHEMA_SOURCE_URL override the upstream URL (default: built from the +# pinned ref below) +# DATA_INGESTORS_REF override the data-ingestors ref (default: the pinned +# SHA in scripts/.data-ingestors-ref, else master) +# SCHEMA_OUT override the in-tree destination (default: internal/schema/ingest.v1.json) +# +# The ref is PINNED (scripts/.data-ingestors-ref), not a floating branch, so an +# unrelated upstream commit doesn't red every open CLI PR — adopting upstream +# is a deliberate SHA bump + re-sync (backend#1009). # # Future: when we cut a v2 schema, this script will need to learn # about multiple versions (e.g. embed v1 AND v2 side-by-side, picked @@ -26,7 +33,28 @@ set -euo pipefail -readonly DEFAULT_URL="https://raw.githubusercontent.com/tracebloc/data-ingestors/master/tracebloc_ingestor/schema/ingest.v1.json" +# The pinned data-ingestors ref: first non-comment, non-blank line of the ref +# file (a full commit SHA), overridable via DATA_INGESTORS_REF, falling back to +# master if the file is somehow absent. +REF_FILE="$(cd "$(dirname "$0")" && pwd)/.data-ingestors-ref" +readonly REF_FILE +_pinned_ref="$(grep -vE '^[[:space:]]*(#|$)' "$REF_FILE" 2>/dev/null | head -1 | tr -d '[:space:]' || true)" +DATA_INGESTORS_REF="${DATA_INGESTORS_REF:-${_pinned_ref:-master}}" + +# The ref is interpolated into a download URL, so validate it before use +# (like scripts/install.sh does for its release tag): a crafted ref — most +# plausibly via the DATA_INGESTORS_REF override — could otherwise inject path +# traversal ("../..") or extra segments into the raw.githubusercontent path. +# Allow only a SHA / branch / tag shape: alnum start, then alnum . _ - / and +# no ".." component. +if ! printf '%s' "$DATA_INGESTORS_REF" | grep -qE '^[A-Za-z0-9][A-Za-z0-9._/-]*$' \ + || printf '%s' "$DATA_INGESTORS_REF" | grep -q '\.\.'; then + echo "error: invalid data-ingestors ref '$DATA_INGESTORS_REF' — expected a commit SHA, branch, or tag" >&2 + echo "(set it in scripts/.data-ingestors-ref or via DATA_INGESTORS_REF)" >&2 + exit 2 +fi + +readonly DEFAULT_URL="https://raw.githubusercontent.com/tracebloc/data-ingestors/${DATA_INGESTORS_REF}/tracebloc_ingestor/schema/ingest.v1.json" readonly DEFAULT_OUT="internal/schema/ingest.v1.json" SCHEMA_SOURCE_URL="${SCHEMA_SOURCE_URL:-$DEFAULT_URL}" diff --git a/scripts/sync-validator-goldens.sh b/scripts/sync-validator-goldens.sh index 8cfc02af..9170e1b0 100755 --- a/scripts/sync-validator-goldens.sh +++ b/scripts/sync-validator-goldens.sh @@ -17,17 +17,21 @@ if [[ "${1:-}" == "--check" ]]; then cp "$GOLDENS" "$tmp/committed.json" "$PYTHON" scripts/gen-validator-goldens.py >/dev/null # Compare VERDICTS only — error text may drift harmlessly (and embeds - # fixture paths); verdicts may not. + # fixture paths); verdicts may not. VALUE-level goldens (resolved label + + # row count + class set) carry no paths, so compare them too — a value-only + # drift (the data-ingestors #340 class: verdict unchanged, stored labels + # change) must fail the check, not slip through. if ! "$PYTHON" -c " import json,sys a=json.load(open('$tmp/committed.json'))['verdicts'] b=json.load(open('$GOLDENS'))['verdicts'] -va={k:v['verdict'] for k,v in a.items()}; vb={k:v['verdict'] for k,v in b.items()} -sys.exit(0 if va==vb else 1) +def view(d): return {k:(v['verdict'], v.get('values')) for k,v in d.items()} +sys.exit(0 if view(a)==view(b) else 1) "; then cp "$tmp/committed.json" "$GOLDENS" # restore — check must not mutate - echo "DRIFT: the ingestor's validator verdicts changed. Re-run the generator," >&2 - echo "commit the new goldens, and update cases.json (+ the Go preview) consciously." >&2 + echo "DRIFT: the ingestor's validator verdicts or read-path VALUES changed. Re-run" >&2 + echo "the generator, commit the new goldens, and update cases.json (+ the Go preview)" >&2 + echo "consciously." >&2 exit 1 fi cp "$tmp/committed.json" "$GOLDENS" # keep the committed copy (paths etc. unchanged)