fix: pipeline annotation false-positives from body comments (#9736)

* fix: reject pipeline `# tag` annotation false-positives on regular comments

`parse_pipeline_annotations` treats any comment line starting with
`# tag <text>` as a worker-tag annotation. In Python scripts, ordinary
English comments beginning with "# tag ..." were misinterpreted: values
over 50 chars failed the `script.tag` INSERT (varchar(50)), and shorter
ones silently overrode the script's worker tag.

Worker tags are single-word identifiers (e.g. `heavy`, `gpu`), so reject
any candidate that contains whitespace or exceeds 50 characters. Mirror
the same validation in the TS parity parser and add regression tests on
both sides.

Fixes WIN-2090

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix: restrict pipeline annotation scan to the leading comment header

The root cause of the `# tag` false-positive is broader than the `tag`
keyword: `parse_pipeline_annotations` scanned every comment line in the
whole file, so any body comment matching an annotation grammar
(`on`, `freshness`, `tag`, `retry`, ...) was misinterpreted. The `tag`
case was the most visible because an over-length value crashed the
`script.tag` INSERT (varchar(50)).

Windmill's other comment-directive parsers (BashAnnotations::sandbox_image,
ssh_target) already scan only the leading comment header and stop at the
first line of real code. Align parse_pipeline_annotations (and its TS
mirror) with that convention: skip blank lines, break on the first
non-comment line. This eliminates body-comment false-positives for every
annotation, not just `tag`.

The `tag` whitespace/length guard from the previous commit is kept as
defense for prose that sits in the header itself.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/parsers/windmill-parser/src/asset_parser.rs

@@ -487,9 +487,13 @@ fn parse_kv_opts(s: &str) -> BTreeMap<String, String> {
     out
 }
 
-// Scan raw source for pipeline annotations. Language-agnostic: any line
-// whose first non-whitespace tokens are a comment prefix (`//`, `#`, or
-// `--`) followed by one of the recognized keywords:
+// Scan the leading comment header for pipeline annotations. Only the
+// contiguous block of comment lines at the top of the file is considered
+// (blank lines tolerated, scan stops at the first line of actual code) so
+// that ordinary comments in the body can't false-positive as annotations.
+// Language-agnostic: any header line whose first non-whitespace tokens are
+// a comment prefix (`//`, `#`, or `--`) followed by one of the recognized
+// keywords:
 //   - `pipeline`                 → opt-in marker (must be alone on the line)
 //   - `on <trigger-spec>`        → asset / native trigger edge (including
 //                                  the marker-only `on schedule` form)
@@ -527,6 +531,9 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
 
     for raw_line in code.lines() {
         let line = raw_line.trim_start();
+        if line.is_empty() {
+            continue;
+        }
         let rest = if let Some(r) = line.strip_prefix("//") {
             r
         } else if let Some(r) = line.strip_prefix("--") {
@@ -534,7 +541,11 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
         } else if let Some(r) = line.strip_prefix('#') {
             r
         } else {
-            continue;
+            // Annotations live in the leading comment header. Stop at the first
+            // line of actual code so comments inside the body (e.g. a regular
+            // `# tag ...` prose comment) can't false-positive as annotations.
+            // Mirrors BashAnnotations::sandbox_image / ssh_target.
+            break;
         };
         let rest = rest.trim_start();
 
@@ -584,7 +595,14 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
 
         if let Some(after_kw) = consume_keyword(rest, "tag") {
             let name = after_kw.trim();
-            if !name.is_empty() && out.tag.is_none() {
+            // Worker tags are single-word identifiers (e.g. `heavy`, `gpu`).
+            // A value with whitespace or beyond the `script.tag` column width
+            // is almost certainly a regular comment starting with "# tag ...".
+            if !name.is_empty()
+                && !name.contains(char::is_whitespace)
+                && name.len() <= 50
+                && out.tag.is_none()
+            {
                 out.tag = Some(name.to_string());
             }
             continue;
@@ -1074,6 +1092,56 @@ mod pipeline_annotation_tests {
         assert!(out.tag.is_none());
     }
 
+    #[test]
+    fn tag_with_whitespace_is_skipped() {
+        // A regular English comment starting with "# tag " must not be
+        // mistaken for a worker-tag annotation (worker tags are single words).
+        let out =
+            parse_pipeline_annotations("# tag this function so we remember to refactor it later");
+        assert!(out.tag.is_none());
+    }
+
+    #[test]
+    fn tag_too_long_is_skipped() {
+        let long = "x".repeat(51);
+        let out = parse_pipeline_annotations(&format!("// tag {long}"));
+        assert!(out.tag.is_none());
+    }
+
+    #[test]
+    fn annotations_in_body_are_ignored() {
+        // Only the leading comment header is scanned. A regular `# tag ...`
+        // prose comment buried in the body — the WIN-2090 false-positive that
+        // crashed the `script.tag` INSERT — must not be treated as an
+        // annotation once real code has started.
+        let code = concat!(
+            "import pandas as pd\n",
+            "\n",
+            "def main():\n",
+            "    # tag each row with its source so downstream steps can filter\n",
+            "    # on s3://should/not/parse\n",
+            "    return pd.DataFrame()\n",
+        );
+        let out = parse_pipeline_annotations(code);
+        assert!(out.tag.is_none());
+        assert!(out.triggers.is_empty());
+    }
+
+    #[test]
+    fn header_allows_blank_lines_before_code() {
+        // Blank lines (e.g. after a shebang) don't end the header; the first
+        // line of real code does.
+        let code = concat!(
+            "#!/usr/bin/env python\n",
+            "\n",
+            "# tag heavy\n",
+            "import os\n",
+            "# tag light\n",
+        );
+        let out = parse_pipeline_annotations(code);
+        assert_eq!(out.tag.as_deref(), Some("heavy"));
+    }
+
     #[test]
     fn retry_count_only() {
         let out = parse_pipeline_annotations("// retry 3");

frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.test.ts

@@ -25,6 +25,38 @@ describe('parsePipelineAnnotations: tag', () => {
 		const out = parsePipelineAnnotations('// tagged heavy')
 		expect(out.tag).toBeUndefined()
 	})
+
+	it('skips a tag value containing whitespace (regular comment false-positive)', () => {
+		const out = parsePipelineAnnotations('# tag this function so we remember to refactor it later')
+		expect(out.tag).toBeUndefined()
+	})
+
+	it('skips a tag value longer than 50 chars', () => {
+		const out = parsePipelineAnnotations('// tag ' + 'x'.repeat(51))
+		expect(out.tag).toBeUndefined()
+	})
+})
+
+describe('parsePipelineAnnotations: header scan', () => {
+	it('ignores annotations in the body once code has started', () => {
+		const code = [
+			'import pandas as pd',
+			'',
+			'def main():',
+			'    # tag each row with its source so downstream steps can filter',
+			'    # on s3://should/not/parse',
+			'    return pd.DataFrame()'
+		].join('\n')
+		const out = parsePipelineAnnotations(code)
+		expect(out.tag).toBeUndefined()
+		expect(out.triggerAssets).toHaveLength(0)
+	})
+
+	it('tolerates blank lines before code but stops at the first code line', () => {
+		const code = ['#!/usr/bin/env python', '', '# tag heavy', 'import os', '# tag light'].join('\n')
+		const out = parsePipelineAnnotations(code)
+		expect(out.tag).toBe('heavy')
+	})
 })
 
 describe('parsePipelineAnnotations: retry', () => {

frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts

@@ -291,8 +291,13 @@ export function parsePipelineAnnotations(code: string): PipelineAnnotations {
 	}
 
 	for (const rawLine of code.split('\n')) {
+		// Annotations live in the leading comment header: skip blank lines but
+		// stop at the first line of actual code, so comments inside the body
+		// (e.g. a regular `# tag ...` prose comment) can't false-positive.
+		// Mirrors the Rust parse_pipeline_annotations header scan.
+		if (rawLine.trim() === '') continue
 		const rest = stripCommentPrefix(rawLine)
-		if (rest === undefined) continue
+		if (rest === undefined) break
 		const inner = rest.trimStart()
 
 		const afterPipeline = consumeKeyword(inner, 'pipeline')
@@ -323,7 +328,10 @@ export function parsePipelineAnnotations(code: string): PipelineAnnotations {
 		const afterTag = consumeKeyword(inner, 'tag')
 		if (afterTag !== undefined) {
 			const name = afterTag.trim()
-			if (name && !out.tag) {
+			// Worker tags are single-word identifiers; a value with whitespace
+			// or beyond the script.tag column width is almost certainly a
+			// regular comment starting with "# tag ...".
+			if (name && !out.tag && !/\s/.test(name) && name.length <= 50) {
 				out.tag = name
 			}
 			continue