fix: harden login env-validation, Inf/NaN schema inference, and log-scanner buffer (bug-hunt MED)

internal/api/client.go

@@ -106,6 +106,21 @@ func ResolveEnv(explicit string) string {
 	return EnvProd
 }
 
+// IsKnownEnv reports whether env is one of the recognized backends (dev/stg/prod,
+// case-insensitively). Callers that let a human PICK the env (e.g. `login`) use it
+// to reject a typo up front — BaseURL deliberately falls unknown values back to
+// prod (a lenient library default), so without this a `--env staging`/`prd` typo
+// would silently target production. Empty is NOT known here: resolve first
+// (ResolveEnv turns empty into the prod default), then validate the result.
+func IsKnownEnv(env string) bool {
+	switch strings.ToLower(env) {
+	case EnvDev, EnvStg, EnvProd:
+ return true
+	default:
+ return false
+	}
+}
+
 // Client talks to the backend REST API. Token (the user token from login) is
 // optional: the device-flow endpoints are unauthenticated; provisioning calls
 // set it.

internal/api/client_test.go

@@ -41,6 +41,23 @@ func TestResolveEnv(t *testing.T) {
 	}
 }
 
+func TestIsKnownEnv(t *testing.T) {
+	known := []string{"dev", "stg", "prod", "DEV", "Prod"} // case-insensitive
+	for _, env := range known {
+ if !IsKnownEnv(env) {
+ t.Errorf("IsKnownEnv(%q) = false, want true", env)
+ }
+	}
+	// Typos and the unknown values BaseURL would silently route to prod
+	// must be rejected so `login` fails instead of persisting a prod session.
+	unknown := []string{"staging", "prd", "production", "development", "", " "}
+	for _, env := range unknown {
+ if IsKnownEnv(env) {
+ t.Errorf("IsKnownEnv(%q) = true, want false", env)
+ }
+	}
+}
+
 func TestRequestDeviceCode(t *testing.T) {
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  if r.URL.Path != "/device/code" || r.Method != http.MethodPost {

internal/cli/auth.go

@@ -56,6 +56,15 @@ func runLogin(ctx context.Context, p *ui.Printer, envFlag string) error {
  return &exitError{code: 1, err: err}
 	}
 	env := api.ResolveEnv(envFlag)
+	// login PICKS the session env and persists it (cfg.CurrentEnv below), so a
+	// typo must fail HERE, not silently resolve to prod. BaseURL's lenient
+	// unknown→prod fallback would otherwise route `--env staging` / `CLIENT_ENV=prd`
+	// to production and store it as the active env for every later command.
+	if !api.IsKnownEnv(env) {
+ return &exitError{code: 1, err: fmt.Errorf(
+ "unknown backend environment %q — valid values are dev, stg, prod (default). "+
+ "Check --env / $CLIENT_ENV", env)}
+	}
 	client := newAPIClient(env)
 	p.Detailf("backend %s — requesting a device code …", client.BaseURL)
 

internal/push/tabular_test.go

@@ -167,6 +167,27 @@ func TestInferSchema(t *testing.T) {
 	}
 }
 
+// TestInferSchema_NonFiniteIsNotFloat: Go's strconv.ParseFloat accepts
+// "Inf"/"Infinity"/"NaN", but the ingestor's FLOAT cast rejects a non-finite
+// value — so a column carrying one must NOT infer FLOAT (that would hand the
+// cluster a schema it only refuses AFTER the upload). The di#349 float grammar
+// (floatRE) pre-screens the token before ParseFloat, so "inf"/"NaN" fall
+// through to VARCHAR and preflight matches what the cluster will accept. No
+// parity-fixture case covers this, so pin it here.
+func TestInferSchema_NonFiniteIsNotFloat(t *testing.T) {
+	dir := t.TempDir()
+	csv := writeFile(t, dir, "data.csv",
+ "reading,note\n1.5,ok\ninf,spike\nNaN,dropout\n")
+	res, err := InferSchema(csv)
+	if err != nil {
+ t.Fatalf("InferSchema: %v", err)
+	}
+	// Longest of 1.5/inf/NaN is 3 runes → VARCHAR(3); the point is it is NOT FLOAT.
+	if got := res.Schema["reading"]; got != "VARCHAR(3)" {
+ t.Errorf("schema[reading] = %q, want VARCHAR(3) (Inf/NaN must not infer FLOAT)", got)
+	}
+}
+
 // TestInferSchema_EmptyColumnIsVarchar1: a column with no non-empty sampled
 // value can't be typed from data; it comes back as VARCHAR(1) (mirroring
 // the ingestor's all-missing rule) and is reported in the Empty list so

internal/submit/watch.go

@@ -463,29 +463,16 @@ func streamPodLogsAndParse(
 	// io.Copy would also work but would buffer chunks at the
 	// transport layer, making the output feel laggy on a fast
 	// ingestion.
-	scanner := bufio.NewScanner(tee)
-	// Default scanner buffer is 64 KB per line — fine for log
-	// lines but bump to 1 MB to handle the (rare) case where a
-	// single ingestion-error stacktrace has a multi-KB Python
-	// traceback line.
-	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
-	for scanner.Scan() {
- line := scanner.Bytes()
- // Print the line + a newline (scanner strips the trailing
- // '\n'). errcheck-friendly: we discard the writer error
- // because the exit code is the customer-facing contract.
- _, _ = out.Write(line)
- _, _ = out.Write([]byte("\n"))
-	}
-	if err := scanner.Err(); err != nil {
- // EOF is normal end-of-stream; other errors (network drop
- // mid-stream, ctx cancel) get propagated.
- if !errors.Is(err, io.EOF) {
- // Flush any buffered partial line before returning so
- // the parser sees content even on mid-line failure.
- parser.FlushLine()
- return parser.Result(), err
- }
+	// Pull the whole stream through the tee (which feeds the summary parser)
+	// while rendering it line-by-line for the customer. An over-long DISPLAY
+	// line (a tqdm '\r'-redraw burst) is drained, not fatal — see
+	// streamDisplayAndParse.
+	if err := streamDisplayAndParse(tee, out, displayLineMax); err != nil {
+ // A genuine mid-stream failure (network drop, ctx cancel). Flush any
+ // buffered partial line so the parser sees content even on a mid-line
+ // failure, then propagate.
+ parser.FlushLine()
+ return parser.Result(), err
 	}
 	// Flush at the end of stream too. A Pod that exited without
 	// a trailing newline on its final stdout write would otherwise
@@ -495,6 +482,49 @@ func streamPodLogsAndParse(
 	return parser.Result(), nil
 }
 
+// displayLineMax caps the per-line DISPLAY buffer (grows on demand from 64 KB,
+// so ordinary log lines still cost 64 KB). A tqdm progress "line" — many '\r'
+// redraws with no '\n' — grows for the life of the run; the generous cap lets
+// the common case render fully, and streamDisplayAndParse drains anything past
+// it rather than failing. The 1h JobWatchTimeout bounds accumulation.
+const displayLineMax = 16 * 1024 * 1024
+
+// streamDisplayAndParse renders r line-by-line to out for the customer. The
+// caller wraps the log stream in an io.TeeReader that ALSO feeds the summary
+// parser, and this is the ONLY place the tee is pulled — so the function's real
+// job is to pull the WHOLE stream through, whatever the display does with it.
+//
+// A tqdm progress "line" (many '\r' redraws, no '\n') can outgrow maxLine. That
+// must NOT stop the pull: if it did, the tee would stop feeding the parser and
+// the closing banner would never be parsed, so watch would report a false exit
+// 9 on a healthy run (raising maxLine only postpones the threshold). On
+// bufio.ErrTooLong we therefore keep draining the rest of the stream —
+// discarding only the oversized DISPLAY line — so the parser still sees the
+// banner; the Job status poll is the verdict's real source of truth. Returns a
+// non-nil error only on a genuine read failure (network drop, ctx cancel).
+func streamDisplayAndParse(r io.Reader, out io.Writer, maxLine int) error {
+	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 0, 64*1024), maxLine)
+	for scanner.Scan() {
+ // scanner strips the trailing '\n'; re-add it. errcheck-friendly: the
+ // write error is discarded because the exit code is the contract.
+ _, _ = out.Write(scanner.Bytes())
+ _, _ = out.Write([]byte("\n"))
+	}
+	err := scanner.Err()
+	if err == nil || errors.Is(err, io.EOF) {
+ return nil
+	}
+	if errors.Is(err, bufio.ErrTooLong) {
+ // Keep draining THROUGH r (the tee) so the parser still sees the banner.
+ if _, derr := io.Copy(io.Discard, r); derr != nil {
+ return derr
+ }
+ return nil
+	}
+	return err
+}
+
 // parserWriter adapts a SummaryParser into an io.Writer for use
 // with io.TeeReader. The TeeReader writes everything to its
 // secondary sink as bytes flow through; the parser's Feed method

internal/submit/watch_display_test.go

@@ -0,0 +1,67 @@
+package submit
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"strings"
+	"testing"
+)
+
+// errAfterReader yields its data, then returns err on the next Read — a genuine
+// mid-stream read failure (network drop / ctx cancel), distinct from io.EOF.
+type errAfterReader struct {
+	data []byte
+	err error
+	pos int
+}
+
+func (r *errAfterReader) Read(p []byte) (int, error) {
+	if r.pos >= len(r.data) {
+ return 0, r.err
+	}
+	n := copy(p, r.data[r.pos:])
+	r.pos += n
+	return n, nil
+}
+
+// TestStreamDisplayAndParse_DrainsPastOversizedLineSoParserSeesBanner pins the
+// #208-review fix: an over-long tqdm-style progress "line" (many '\r' redraws,
+// no '\n') that outgrows the display buffer, immediately followed by the real
+// closing banner. Because the tee is pulled ONLY by the display scanner, a
+// naive ErrTooLong bail would stop the parser from ever seeing the banner →
+// watch returns a false exit 9 on a healthy run. The drain-past-ErrTooLong must
+// keep pulling so the parser still resolves the summary.
+func TestStreamDisplayAndParse_DrainsPastOversizedLineSoParserSeesBanner(t *testing.T) {
+	oversized := strings.Repeat("\rprocessing... ", 500) // ~7 KB, no '\n'
+	stream := strings.NewReader(oversized + realIngestorBanner)
+	parser := NewSummaryParser()
+	tee := io.TeeReader(stream, parserWriter{parser: parser})
+
+	var out bytes.Buffer
+	// Tiny cap so the oversized line trips ErrTooLong (production uses 16 MB).
+	if err := streamDisplayAndParse(tee, &out, 1024); err != nil {
+ t.Fatalf("an over-long DISPLAY line must not be fatal; got: %v", err)
+	}
+	parser.FlushLine()
+	if got := parser.Result().InsertedRecords; got != 1200 {
+ t.Fatalf(
+ "parser missed the banner after the oversized line (false exit 9): "+
+ "InsertedRecords=%d, want 1200",
+ got,
+ )
+	}
+}
+
+// TestStreamDisplayAndParse_GenuineReadErrorPropagates: a real mid-stream read
+// failure (not ErrTooLong, not EOF) stays fatal — the drain path must not
+// swallow genuine stream errors.
+func TestStreamDisplayAndParse_GenuineReadErrorPropagates(t *testing.T) {
+	want := errors.New("connection reset by peer")
+	r := &errAfterReader{data: []byte("some log line\n"), err: want}
+
+	var out bytes.Buffer
+	if err := streamDisplayAndParse(r, &out, 1024); !errors.Is(err, want) {
+ t.Fatalf("a genuine read error should propagate; got: %v", err)
+	}
+}