${esc(d.field_count)} fields analyzed
Selected
Entity: ${esc((selected.entity || []).join(', ') || '-')}
Time: ${esc((selected.time || []).join(', ') || '-')}
Categorical: ${esc((selected.categorical || []).join(', ') || '-')}
Numeric: ${esc((selected.numeric || []).join(', ') || '-')}
Top fields
${esc(top || '-')}
Rejected
${esc(rejected || '-')}
${esc(reasons || '')}
`;
+ return `${esc(d.field_count)} fields analyzed
Selected
Entity: ${esc((selected.entity || []).join(', ') || '-')}
Time: ${esc((selected.time || []).join(', ') || '-')}
Categorical: ${esc((selected.categorical || []).join(', ') || '-')}
Numeric: ${esc((selected.numeric || []).join(', ') || '-')}
Shared across streams
${esc(shared || '-')}
Top fields
${esc(top || '-')}
Rejected
${esc(rejected || '-')}
${esc(reasons || '')}
`;
}
function drawTrend(history) {
const canvas=document.getElementById('trendChart'), ctx=canvas.getContext('2d'), ratio=window.devicePixelRatio||1, cw=canvas.clientWidth, ch=canvas.clientHeight;
diff --git a/src/fgai/llm.py b/src/fgai/llm.py
index f69307c..1f89679 100644
--- a/src/fgai/llm.py
+++ b/src/fgai/llm.py
@@ -94,6 +94,8 @@ def ollama_profile_advice(suggestions: list[dict[str, object]], model: str | Non
"stream_name": item.get("stream_name"),
"events": item.get("events"),
"common_fields": item.get("common_fields", [])[:20],
+ "shared_fields": item.get("shared_fields", [])[:12],
+ "detected_log_type": item.get("detected_log_type", ""),
"heuristic_profile": item.get("profile", {}),
}
for item in suggestions[:10]
@@ -110,7 +112,7 @@ def ollama_profile_advice(suggestions: list[dict[str, object]], model: str | Non
"{\"profiles\":[{\"stream_id\":\"...\",\"entity_fields\":[\"...\"],\"timestamp_field\":\"...\","
"\"categorical_fields\":[\"...\"],\"numeric_fields\":[\"...\"],\"detectors\":{\"auth_failure\":{\"enabled\":true,\"minimum\":5,\"z_threshold\":3}},"
"\"reason\":\"short reason\"}]}. "
- "Use only field names present in common_fields or heuristic_profile. Do not include raw message/full_message fields. "
+ "Use only field names present in common_fields, shared_fields, or heuristic_profile. Prefer fields that appear in shared_fields when they are useful categorical or numeric baseline fields. Do not include raw message/full_message fields. "
"Allowed detectors are auth_failure, dns_query, deny_action. Prefer canonical fields such as username, hostname, eventid, srcip, dstip when present. "
f"\n\nObserved streams:\n{json.dumps(compact, sort_keys=True)}"
),
diff --git a/src/fgai/profile_suggestions.py b/src/fgai/profile_suggestions.py
index 1566631..89198b9 100644
--- a/src/fgai/profile_suggestions.py
+++ b/src/fgai/profile_suggestions.py
@@ -359,6 +359,35 @@ def _generic_numeric_fields(coverage: Counter[str], numeric_counts: Counter[str]
return output
+def _shared_profile_fields(
+ coverage: Counter[str],
+ unique_values: dict[str, set[str]],
+ numeric_counts: Counter[str],
+ total: int,
+ stream_field_counts: Counter[str],
+) -> tuple[list[str], list[str], list[dict[str, object]]]:
+ categorical: list[str] = []
+ numeric: list[str] = []
+ stats: list[dict[str, object]] = []
+ for field, stream_count in stream_field_counts.most_common():
+ if stream_count < 2 or field in IGNORED_DISCOVERY_FIELDS or not coverage[field]:
+ continue
+ coverage_ratio = coverage[field] / max(1, total)
+ cardinality = len(unique_values[field])
+ if coverage_ratio < 0.1 or cardinality <= 1:
+ continue
+ if cardinality > min(200, max(8, int(total * 0.7))):
+ continue
+ numeric_ratio = numeric_counts[field] / max(1, coverage[field])
+ row = {"field": field, "streams": stream_count, "coverage": round(coverage_ratio, 2), "unique_values": cardinality}
+ stats.append(row)
+ if numeric_ratio >= 0.95:
+ numeric.append(field)
+ elif not _looks_like_entity_field(field) and not _looks_like_time_field(field):
+ categorical.append(field)
+ return categorical[:8], numeric[:5], stats[:12]
+
+
def _allowed_fields(suggestion: dict[str, object]) -> set[str]:
profile = suggestion.get("profile", {}) if isinstance(suggestion.get("profile"), dict) else {}
fields = {str(item.get("field", "")) for item in suggestion.get("common_fields", []) if isinstance(item, dict)}
@@ -432,6 +461,15 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
stream_id = _stream_id(event)
grouped[stream_id].append(event)
names.setdefault(stream_id, _stream_name(event, stream_id))
+ stream_field_counts: Counter[str] = Counter()
+ for stream_events in grouped.values():
+ fields = {
+ field
+ for event in stream_events
+ for field, value in event.fields.items()
+ if field not in IGNORED_DISCOVERY_FIELDS and _has_discovery_value(value)
+ }
+ stream_field_counts.update(fields)
suggestions: list[dict[str, object]] = []
for stream_id, stream_events in grouped.items():
@@ -457,15 +495,18 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
entity_priority = WINDOWS_ENTITY_PRIORITY if is_windows else ENTITY_PRIORITY
categorical_priority = WINDOWS_CATEGORICAL_PRIORITY if is_windows else CATEGORICAL_PRIORITY
numeric_priority = WINDOWS_NUMERIC_PRIORITY if is_windows else NUMERIC_PRIORITY
+ shared_categorical, shared_numeric, shared_stats = _shared_profile_fields(coverage, unique_values, numeric_counts, total, stream_field_counts)
entity_fields = _generic_entity_fields(coverage, unique_values, total, _pick_present(entity_priority, coverage, total, min_ratio=0.02, limit=12 if is_windows else 3))
timestamp = next(iter(_pick_present(TIME_PRIORITY, coverage, total, min_ratio=0.02, limit=1)), "")
if not timestamp:
timestamp = next((field for field, count in coverage.most_common() if _looks_like_time_field(field) and count / max(1, total) >= 0.02), "timestamp")
- categorical = _generic_categorical_fields(coverage, unique_values, total, _pick_present(categorical_priority, coverage, total, min_ratio=0.02, limit=10 if is_windows else 8))
+ categorical_seed = list(dict.fromkeys([*_pick_present(categorical_priority, coverage, total, min_ratio=0.02, limit=10 if is_windows else 8), *shared_categorical]))
+ categorical = _generic_categorical_fields(coverage, unique_values, total, categorical_seed)
numeric_seed = [
field for field in numeric_priority
if coverage[field] and numeric_counts[field] / max(1, coverage[field]) >= 0.8
][:5]
+ numeric_seed = list(dict.fromkeys([*numeric_seed, *shared_numeric]))
numeric = numeric_seed if is_windows else _generic_numeric_fields(coverage, numeric_counts, total, numeric_seed)
if not entity_fields:
entity_fields = [field for field, _count in coverage.most_common() if field not in IGNORED_DISCOVERY_FIELDS][:1]
@@ -522,10 +563,12 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
"numeric": numeric,
},
"top_fields": _field_stats(coverage, unique_values, numeric_counts, total)[:16],
+ "shared_fields": shared_stats,
"rejected_fields": _rejected_fields(coverage, unique_values, numeric_counts, total, selected_fields),
"reasons": [
f"detected {log_type} log pattern",
f"selected {len(entity_fields)} entity field(s), {len(categorical)} categorical field(s), and {len(numeric)} numeric field(s)",
+ f"found {len(shared_stats)} field(s) that also appear in other enabled streams",
"ignored raw/internal, constant, sparse, and very high-cardinality fields",
],
}
@@ -544,6 +587,7 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
"numeric_fields": numeric,
"detectors": detectors,
"common_fields": high_coverage,
+ "shared_fields": shared_stats,
"discovery": discovery,
"detected_log_type": log_type,
"profile": {
diff --git a/tests/test_profile_suggestions.py b/tests/test_profile_suggestions.py
index 017f8f7..c176011 100644
--- a/tests/test_profile_suggestions.py
+++ b/tests/test_profile_suggestions.py
@@ -154,6 +154,30 @@ class ProfileSuggestionTests(unittest.TestCase):
self.assertEqual(rejected["constant_field"], "constant value")
self.assertEqual(rejected["request_id"], "too high cardinality")
+ def test_fields_shared_between_streams_are_default_profile_candidates(self):
+ events = []
+ for stream_id, stream_name in (("billing", "BillingApp"), ("orders", "OrdersApp")):
+ events.extend(
+ parse_log_line(
+ f"fgai_stream_id={stream_id} fgai_stream={stream_name} actor_id=user{index % 6} "
+ f"tenant_id=t{index % 3} workflow_state={'approved' if index % 2 else 'rejected'} "
+ f"result_code=r{index % 4} latency_ms={index % 9} created_at=2026-06-29T10:00:{index:02d}Z"
+ )
+ for index in range(1, 30)
+ )
+
+ suggestions = {item["stream_id"]: item for item in suggest_stream_profiles(events)}
+
+ for suggestion in suggestions.values():
+ profile = suggestion["profile"]
+ shared = {item["field"] for item in suggestion["shared_fields"]}
+ self.assertIn("workflow_state", shared)
+ self.assertIn("result_code", shared)
+ self.assertIn("latency_ms", shared)
+ self.assertIn("workflow_state", profile["categorical_fields"])
+ self.assertIn("result_code", profile["categorical_fields"])
+ self.assertIn("latency_ms", profile["numeric_fields"])
+
def test_applies_valid_llm_advice_and_rejects_unknown_fields(self):
suggestion = suggest_stream_profiles([
parse_log_line("fgai_stream_id=windows fgai_stream=Windows username=alice hostname=host01 eventid=4625 action=failure")