fix: resolve PR #219 conflicts and address CladeCumulus & validated-statistics findings

.github/workflows/ci.yml

@@ -400,11 +400,30 @@ jobs:
         shell: bash
         run: |
           set +e
-          julia --project=. -t2 --compiled-modules=no test/runtests.jl numeric_policy exact_summaries high_precision derived_analyses validated_statistics 2>&1 | tee /tmp/julia-validated-tests.log
+          julia --project=. -t2 --compiled-modules=no test/runtests.jl numeric_policy exact_summaries high_precision derived_analyses validated_statistics 2>&1 | tee /tmp/validated-statistics-tests.log
           status="${PIPESTATUS[0]}"
           if [ "$status" -ne 0 ]; then
             enc() { head -c 3500 | sed -e 's/%/%25/g' -e 's/\r/%0D/g' | awk '{printf "%s%%0A", $0}'; }
-            echo "::error title=Validated statistics failed (tail)::$(tail -c 3500 /tmp/julia-validated-tests.log | enc)"
+            awk '
+              /Test Summary:/ { left = 0 }
+              /(Error: (validated run statistics|numeric input did not fail closed|exponent numeric token did not fail closed)|Test Failed at|Error During Test at)/ || (left == 0 && /Got exception outside of a @test/) { n++; left = 60 }
+              left > 0 { print > ("/tmp/validated-failure-" n ".txt"); left-- }
+            ' /tmp/validated-statistics-tests.log
+            shown=0
+            for f in $(ls /tmp/validated-failure-*.txt 2>/dev/null | sort -t- -k3 -n); do
+              shown=$((shown + 1))
+              [ "$shown" -gt 8 ] && break
+              echo "::error title=Validated-statistics failure ${shown}::$(enc < "$f")"
+            done
+            awk '
+              /Test Summary:/ { print; p = index($0, "Pass"); pe = (p ? p + 3 : 0); t = index($0, "Total"); next }
+              pe && t && index($0, "|") && substr($0, pe + 1, t - pe - 1) ~ /[0-9]/ { print; next }
+              /ERROR:|Some tests did not pass/ { print }
+            ' /tmp/validated-statistics-tests.log | tail -n 60 > /tmp/validated-statistics-summary.txt
+            echo "::error title=Validated-statistics tests failed (summary)::$(enc < /tmp/validated-statistics-summary.txt)"
+            if [ "$shown" -eq 0 ]; then
+              echo "::error title=Validated-statistics tests failed (tail)::$(tail -c 3500 /tmp/validated-statistics-tests.log | enc)"
+            fi
           fi
           exit "$status"
 

src/analysis/clade_cumulus.jl

@@ -1029,8 +1029,8 @@ function _graft_jplace_queries(
 
     function walk(u::PhyloNode)::PhyloNode
         rebuilt_kids = PhyloNode[walk(ch) for ch in u.children]
-        cur = PhyloNode(u.id, u.name, u.branch_length, u.support, u.edge_num, rebuilt_kids)
-        specs = u.edge_num === nothing ? nothing : get(by_edge, u.edge_num, nothing)
+        cur = PhyloNode(u.id, u.name, u.branch_length, u.support, u.edge_number, rebuilt_kids)
+        specs = u.edge_number === nothing ? nothing : get(by_edge, u.edge_number, nothing)
         (specs === nothing || isempty(specs)) && return cur
 
         # Order grafts from most distal (closest to `cur`, smallest `distal`) to most proximal.
@@ -1048,7 +1048,7 @@ function _graft_jplace_queries(
             sub = PhyloNode(string("jplace_clade#", seq_auto[]), "", 0.0, 1.0, nothing, PhyloNode[sub, qtip])
         end
         rem_len = max(0.0, total_len - prev_distal)
-        PhyloNode(sub.id, sub.name, rem_len, cur.support, cur.edge_num, sub.children)
+        PhyloNode(sub.id, sub.name, rem_len, cur.support, cur.edge_number, sub.children)
     end
 
     walk(base_root)
@@ -1791,11 +1791,12 @@ function _reaggregate_after_move(
                 mono = get(total_tips_per_taxon, u.dominant_taxon, 0) == 1
             end
             active_here = u.cumulative_count > 0.0
+            sub_pd = active_here && !is_root ? max(0.0, u.branch_length) : 0.0
             st = Epistemic.epistemic_status(case_obj, u.id)
             leaf_node = CladeNode(
                 u.id, u.name, u.rank, u.direct_count, u.cumulative_count, freq,
                 u.residual_count, Epistemic.cloud_size_by_residual(u.residual_count),
-                u.branch_length, u.support, 0.0, u.dominant_taxon, u.taxon_purity, mono,
+                u.branch_length, u.support, sub_pd, u.dominant_taxon, u.taxon_purity, mono,
                 st, Epistemic.epistemic_colour(st), CladeNode[],
             )
             return (leaf_node, tmap, 1, active_here)
@@ -1817,13 +1818,17 @@ function _reaggregate_after_move(
             n_sub_tips += nt
             if ch_active
                 any_active = true
-                sub_pd += max(0.0, rk.branch_length) + rk.subtree_pd
+                sub_pd += rk.subtree_pd
             end
             for (k, v) in ct
                 merged_taxa[k] = get(merged_taxa, k, 0) + v
             end
         end
 
+        # Match `clade_faith_pd`: an active non-root clade contributes its own
+        # incoming branch in addition to the branches accumulated below it.
+        !is_root && any_active && (sub_pd += max(0.0, u.branch_length))
+
         dom_tx = nothing
         dom_n = 0
         for (tx, c) in merged_taxa

src/server/routes/clade_cumulus.jl

@@ -75,19 +75,14 @@ function _load_run_table_df(study::AbstractString, run::AbstractString, table::A
     db_path = joinpath(dir, "results.duckdb")
     isfile(db_path) || return (nothing, nothing, json_error(404, "no_results",
         "No results database for run '$run'"))
-    db = DuckDB.DB(db_path)
-    con = DBInterface.connect(db)
-    try
+    with_results_db(dir) do con
         cols = _duckdb_columns(con, table)
         isempty(cols) && return (nothing, nothing, json_error(404, "table_not_found",
             "Table '$table' not found in run '$run'"))
         scols = _sample_count_columns(con, table)
         order = _stable_order_clause(cols)
         df = DataFrame(DBInterface.execute(con, "SELECT * FROM \"$table\" $order"))
-        return (df, (db_path = db_path, sample_cols = scols, columns = cols), nothing)
-    finally
-        DBInterface.close!(con)
-        close(db)
+        (df, (db_path = db_path, sample_cols = scols, columns = cols), nothing)
     end
 end
 
@@ -97,7 +92,7 @@ function _resolve_tree_inputs_from_request(
     source_digests::OrderedDict{String,String},
 )
     tree_source = lowercase(strip(string(get(body, :treeSource, "auto"))))
-    tree_file = strip(string(get(body, :treeFile, "")))
+    tree_file = String(strip(string(get(body, :treeFile, ""))))
     inline_newick = strip(string(get(body, :newick, "")))
 
     if tree_source == "study_tree" || (!isempty(tree_file) && tree_source == "auto")
@@ -142,10 +137,9 @@ end
     dir = _require_duckdb(study, run; group)
     db_path = isnothing(dir) ? nothing : joinpath(dir, "results.duckdb")
     tables_info = OrderedDict{String,Any}[]
-    if !isnothing(db_path) && isfile(db_path)
-        db = DuckDB.DB(db_path)
-        con = DBInterface.connect(db)
-        try
+    if !isnothing(dir) && !isnothing(db_path) && isfile(db_path)
+        tables_info = with_results_db(dir) do con
+            available = OrderedDict{String,Any}[]
             # Real tables only: SHOW TABLES also lists views, and a view backed
             # by a session-registered DataFrame cannot be bound (let alone
             # counted) from this fresh connection.
@@ -156,20 +150,19 @@ end
                 cols = _duckdb_columns(con, t)
                 scols = _sample_count_columns(con, t)
                 isempty(scols) && continue
-                nrows = Int(only(DataFrame(DBInterface.execute(con, "SELECT COUNT(*) AS n FROM \"$t\""))).n)
+                nrows = Int(only(DataFrame(DBInterface.execute(con,
+                    "SELECT COUNT(*) AS n FROM \"$t\""))).n)
                 has_seq = any(c -> c in ("Sequence", "sequence", "seq", "DNA"), cols)
                 ranks = [r for r in MetaManifold.CladeCumulus.DEFAULT_TAXONOMIC_RANKS if r in cols]
-                push!(tables_info, OrderedDict{String,Any}(
+                push!(available, OrderedDict{String,Any}(
                     "table"        => t,
                     "row_count"    => nrows,
                     "sample_count" => length(scols),
                     "has_sequence" => has_seq,
                     "ranks"        => ranks,
                 ))
             end
-        finally
-            DBInterface.close!(con)
-            close(db)
+            available
         end
     end
 
@@ -208,7 +201,7 @@ end
         return json_error(400, "invalid_json", "Request body must be valid JSON")
     end
 
-    table = strip(string(get(body, :table, "merged")))
+    table = String(strip(string(get(body, :table, "merged"))))
     _valid_name(table) || return json_error(400, "invalid_table", "Invalid table name '$table'")
 
     mode_str = lowercase(strip(string(get(body, :mode, "combined"))))
@@ -276,13 +269,13 @@ end
             jd_div = MetaManifold.CladeCumulus.parse_jplace(jplace_str)
             for q in jd_div.queries
                 q.status === :valid && !isempty(q.candidates) || continue
-                tw = sum((c.weight for c in q.candidates); init = 0.0)
+                tw = sum((c.like_weight for c in q.candidates); init = 0.0)
                 tw > 0.0 || continue
                 multi = length(q.candidates) > 1
                 for (qname, _) in q.names
                     for (ci, cand) in enumerate(q.candidates)
                         tid = multi ? string(qname, "@edge", cand.edge_num, "#", ci) : qname
-                        tip_weights[tid] = cand.weight / tw
+                        tip_weights[tid] = cand.like_weight / tw
                     end
                 end
             end
@@ -399,7 +392,7 @@ end
     (isempty(source_id) || isempty(target_id)) &&
         return json_error(400, "missing_clade_ids", "Both sourceId and targetId are required")
 
-    table = strip(string(get(body, :table, "merged")))
+    table = String(strip(string(get(body, :table, "merged"))))
     _valid_name(table) || return json_error(400, "invalid_table", "Invalid table name '$table'")
 
     mode_str = lowercase(strip(string(get(body, :mode, "combined"))))

src/server/routes/validated_statistics.jl

@@ -41,8 +41,10 @@ end
 function _statistics_body(req)
     length(req.body) <= _STATISTICS_REQUEST_LIMIT || throw(ResourceLimitError(
         "request bytes", _STATISTICS_REQUEST_LIMIT, length(req.body)))
+    raw = String(req.body)
+    _reject_float_count_tokens(raw)
     parsed = try
-        JSON3.read(String(req.body))
+        JSON3.read(raw)
     catch
         throw(ArgumentError("request body is not valid JSON"))
     end
@@ -310,8 +312,9 @@ function _validated_integer_sample_columns(con, table::String, columns::Vector{S
     samples
 end
 
-"""Read and aggregate an integer-safe run table. Only GROUP BY is done in SQL;
-all count addition is Julia BigInt. No SQL SUM and no Float64 matrix is used."""
+"""Read and aggregate an integer-safe run table. SQL only selects bounded rows;
+feature grouping and all count addition happen in Julia BigInt. No SQL SUM or
+Float64 matrix is used."""
 function _reject_float_count_filters(sample_columns::Vector{String}, params::Dict{String,String})
     for sample in sample_columns, prefix in ("col_min.", "col_max.")
         key = prefix * sample
@@ -348,28 +351,29 @@ function _validated_duckdb_counts(con, table::String, columns::Vector{String};
         "COALESCE(NULLIF(CAST($(_sql_identifier(taxon)) AS VARCHAR), ''), 'Unassigned')"
     end
     count_columns = join([_sql_identifier(c) for c in samples], ", ")
-    # Rows sharing a feature (e.g. several sequences in one genus) are summed
-    # here, in BigInt, not in SQL: DuckDB refuses bare count columns beside
-    # GROUP BY, and SUM over HUGEINT counts would overflow exactly the values
-    # this path exists to keep exact. ORDER BY keeps features sorted and each
-    # feature's rows adjacent.
+    # Bound the raw rows before materialising the result. SQL deliberately does
+    # no SUM: feature-level addition is performed below with BigInt so HUGEINT
+    # values cannot overflow while duplicate taxonomy rows are aggregated.
+    max_rows = div(MAX_CELLS, length(samples))
+    count_sql = "SELECT COUNT(*) AS n FROM $(_sql_identifier(table)) " * where_clause
+    row_count = Int(only(DataFrame(DBInterface.execute(con, count_sql, where_params))).n)
+    if row_count > max_rows
+        observed_cells = Int(min(BigInt(row_count) * length(samples), BigInt(typemax(Int))))
+        throw(ResourceLimitError("input table cells", MAX_CELLS, observed_cells))
+    end
+    row_count > 0 || throw(ArgumentError("no rows remain after filtering"))
     sql = "SELECT $feature_expression AS feature, $count_columns FROM $(_sql_identifier(table)) " *
-          "$where_clause ORDER BY feature LIMIT $(MAX_CELLS + 1)"
+          "$where_clause ORDER BY feature LIMIT $max_rows"
     frame = DataFrame(DBInterface.execute(con, sql, where_params))
-    nrow(frame) <= MAX_CELLS || throw(ResourceLimitError("rows", MAX_CELLS, nrow(frame)))
-    nrow(frame) > 0 || throw(ArgumentError("no rows remain after filtering"))
 
-    features = String[]
-    feature_rows = Vector{BigInt}[]
+    aggregated = Dict{String,Vector{BigInt}}()
     for (i, row) in enumerate(eachrow(frame))
         ismissing(row.feature) && throw(UnsupportedRepresentationError(
             "a non-missing feature label", "missing", "validated run statistics row $i"))
         feature = string(row.feature)
-        if isempty(features) || features[end] != feature
-            push!(features, feature)
-            push!(feature_rows, zeros(BigInt, length(samples)))
+        feature_counts = get!(aggregated, feature) do
+            zeros(BigInt, length(samples))
         end
-        target = feature_rows[end]
         for (j, sample) in enumerate(samples)
             value = row[Symbol(sample)]
             ismissing(value) && throw(UnsupportedRepresentationError(
@@ -378,15 +382,15 @@ function _validated_duckdb_counts(con, table::String, columns::Vector{String};
                 "an integer count", repr(value), "feature '$feature', sample '$sample'"))
             value >= 0 || throw(ArgumentError(
                 "negative count $value at feature '$feature', sample '$sample'"))
-            target[j] += BigInt(value) # every addition is exact BigInt
+            feature_counts[j] += BigInt(value) # every addition is exact BigInt
         end
     end
+
+    features = sort!(collect(keys(aggregated)))
     counts = Matrix{BigInt}(undef, length(features), length(samples))
-    for (i, feature_row) in enumerate(feature_rows)
-        counts[i, :] = feature_row
+    for (i, feature) in enumerate(features)
+        counts[i, :] = aggregated[feature]
     end
-    length(features) * length(samples) <= MAX_CELLS || throw(ResourceLimitError(
-        "table cells", MAX_CELLS, length(features) * length(samples)))
     ExactSummaries.validate_coordinates(counts, samples, features)
     counts, samples, features
 end
@@ -425,7 +429,6 @@ end
 @post "/api/v1/statistics/summaries" function(req)
     _statistics_guard() do
         body = _statistics_body(req)
-        _reject_float_count_tokens(String(req.body))
         policy = _statistics_policy(body)
         counts, labels, features, groups = _direct_count_input(body)
         result = _run_statistics(counts, labels, features, groups, policy)

test/unit/test_clade_cumulus.jl

@@ -14,7 +14,9 @@ const EP = MetaManifold.Epistemic
 const CC = MetaManifold.CladeCumulus
 const DM = MetaManifold.DiversityMetrics
 const PH = MetaManifold.Phylogeny
-const SV = MetaManifold.Server
+# Not const: other test files bind SV in Main the same way, and runtests.jl
+# includes them all into one module.
+SV = MetaManifold.Server
 
 _cc_req(method::String, path::String, body::String = "") =
     SV.Oxygen.internalrequest(
@@ -134,7 +136,7 @@ _cc_req(method::String, path::String, body::String = "") =
             "placements" => [
                 OrderedDict("n" => ["Q_split"], "p" => [[0, -100.0, 0.60, 0.01, 0.02], [1, -100.4, 0.40, 0.01, 0.03]]),
                 OrderedDict("nm" => [["Q_borderline", 1]], "p" => [[3, -98.0, 0.85, 0.05, 0.02], [2, -101.0, 0.15, 0.05, 0.02]]),
-                OrderedDict("n" => ["Q_strong"], "p" => [[3, -97.0, 0.98, 0.02, 0.01]]),
+                OrderedDict("n" => ["Q_strong"], "p" => [[3, -97.0, 0.98, 0.02, 0.01], [2, -101.0, 0.02, 0.02, 0.01]]),
                 OrderedDict("n" => ["Q_bad_sum"], "p" => [[0, -100.0, 0.40, 0.01, 0.01]]),
             ],
         ))
@@ -275,6 +277,31 @@ echo "(MMTIP_000001:0.05,(MMTIP_000002:0.03,MMTIP_000003:0.04)0.910:0.12,MMTIP_0
         end
     end
 
+    @testset "Nested-clade drag-and-drop counts each active branch once" begin
+        df = DataFrame(
+            SeqName = ["ASV_1", "ASV_2", "ASV_3", "ASV_4", "ASV_5", "ASV_6"],
+            S1 = [10, 20, 30, 40, 50, 0],
+        )
+        tree = CC.build_clade_tree(df; mode = :phylogenetic, newick =
+            "((ASV_1:0.10,ASV_2:0.20,ASV_6:9.0)0.96:0.30,(ASV_3:0.40,ASV_4:0.50,ASV_5:0.70)0.97:0.60)Root:8.0;")
+        source, target = tree.root.children
+        verdict = CC.validate_drag_drop(tree, source.id, target.id)
+        @test verdict["allowed"]
+        proposed = verdict["proposed_root"]
+        target_after = only(proposed["children"])
+        source_after = last(target_after["children"])
+
+        # 0.10 + 0.20 + 0.30 + 0.40 + 0.50 + 0.70 + 0.60 = 2.80.
+        # The inactive tip's 9.0 and the root's 8.0 contribute nothing.
+        @test proposed["subtree_pd"] ≈ 2.80 atol = 1e-12
+        @test target_after["subtree_pd"] ≈ 2.80 atol = 1e-12
+        @test source_after["subtree_pd"] ≈ 0.60 atol = 1e-12
+        @test [ch["subtree_pd"] for ch in source_after["children"]] ≈ [0.10, 0.20, 0.0] atol = 1e-12
+        @test proposed["cumulative_count"] == tree.root.cumulative_count == 150.0
+        @test tree.root.subtree_pd ≈ 2.80 atol = 1e-12
+        @test length(tree.root.children) == 2
+    end
+
     @testset "Dual-hierarchy CladeCumulus, polyphyly detection, drag-and-drop, and HTTP routes" begin
         tmp = mktempdir()
         old_root = SV.ServerState._root[]
@@ -308,9 +335,16 @@ echo "(MMTIP_000001:0.05,(MMTIP_000002:0.03,MMTIP_000003:0.04)0.910:0.12,MMTIP_0
             db_path = joinpath(merge_dir, "results.duckdb")
             db = DuckDB.DB(db_path)
             con = DBInterface.connect(db)
-            DuckDB.register_data_frame(con, df, "src_df")
-            DBInterface.execute(con, "CREATE TABLE merged AS SELECT * FROM src_df")
+            DBInterface.execute(con, """CREATE TABLE merged (
+                SeqName VARCHAR, Domain VARCHAR, Genus VARCHAR, Sequence VARCHAR,
+                SampleA BIGINT, SampleB BIGINT)""")
+            for row in eachrow(df)
+                DBInterface.execute(con, "INSERT INTO merged VALUES (?, ?, ?, ?, ?, ?)",
+                    [row.SeqName, row.Domain, row.Genus, row.Sequence, row.SampleA, row.SampleB])
+            end
+            DBInterface.execute(con, "CHECKPOINT")
             DBInterface.close!(con)
+            DBInterface.close!(db)
             close(db)
             before_db_sha = bytes2hex(open(sha256, db_path))